"""Merging object tables, aggregating each measurement by what it MEASURES.
A cell with four pathogens in it has four rows in ``pathogen`` and one in
``cell``. Putting a pathogen measurement on the same axis as a cell
measurement means rolling those four up into one number -- and which number
depends entirely on what is being measured:
area, perimeter, integrated intensity, counts -> SUM
minimum intensity -> MIN of the four
maximum intensity -> MAX of the four
mean, median -> the object mean/median
shape descriptors, positions -> mean
text -> the first
``spacr.io._read_and_join_tables`` already does this join, and aggregates
every numeric column with ``mean``. That silently answers a different question
per column: four pathogens' total area becomes an average area, a count
becomes an average count, and a MINIMUM becomes a mean of minima, which is not
a minimum of anything. The join is otherwise sound -- it is where the parent
link, the timelapse key and the cardinality checks live -- so this module
changes what the aggregation is, not how the tables find each other.
**Naming.** A merged column carries its table: ``area`` from ``nucleus``
becomes ``nucleus_area``. Prefix rather than suffix so every column of one
object sorts together in the axis picker, which is how anyone looks for them.
**The primary object.** Everything is rolled up onto ONE table -- the cell by
default. That choice decides what a row means, so it is a setting rather than
an assumption: rolling cells onto pathogens is a legitimate thing to want and
gives a different table.
"""
from __future__ import annotations
import logging
import re
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Mapping, Optional, Sequence, Tuple
import numpy as np
import pandas as pd
from .object_roles import (
ANCHOR_COLUMN,
ORGANELLE_ROLES,
anchor_column,
is_one_row_per_cell,
)
LOG = logging.getLogger("spacr.merge_tables")
#: How a child's rows combine into one number for the parent.
SUM, MIN, MAX, MEAN, MEDIAN, FIRST = "sum", "min", "max", "mean", "median", "first"
#: The aggregations offered, for a settings dropdown.
AGGREGATIONS: Tuple[str, ...] = (SUM, MIN, MAX, MEAN, MEDIAN, FIRST)
#: Additional explicit methods offered by the typed custom merge editor.
EXPLICIT_AGGREGATIONS: Tuple[str, ...] = AGGREGATIONS + ("last", "count", "nunique", "any", "all")
#: Column-name patterns -> aggregation, FIRST MATCH WINS, so the order is the
#: rule. Anything unmatched falls to :data:`DEFAULT_AGGREGATION`.
#:
#: Read this as a statement about measurements, not about strings: each entry
#: is here because combining that quantity any other way answers a different
#: question. `min_intensity` is the clearest -- a mean of four minima is not
#: the minimum of anything.
AGGREGATION_RULES: Tuple[Tuple[str, str], ...] = (
(r"(^|_)(object_label|label|id|cell_id|nucleus_id|pathogen_id|"
r"organelle_id|cytoplasm_id|parent_id|prcfo|prcf|prc)(_|$)", FIRST),
(r"(^|_)(count|n_objects|number)(_|$)", SUM),
(r"(^|_)min(imum)?(_|$)", MIN),
(r"(^|_)max(imum)?(_|$)", MAX),
(r"(^|_)median(_|$)", MEDIAN),
(r"(^|_)(mean|average|avg)(_|$)", MEAN),
(r"(^|_)(area|volume|convex_area|filled_area)(_|$)", SUM),
(r"(^|_)(perimeter|length|width|height|diameter|"
r"equivalent_diameter|major_axis_length|minor_axis_length)(_|$)", MEAN),
(r"(^|_)(integrated|total|sum|integral)(_|$)", SUM),
(r"(^|_)(std|stdev|var|variance|mad|iqr|percentile|quantile|"
r"skew|kurtosis|entropy)(_|$)", MEAN),
(r"(^|_)(eccentricity|solidity|extent|circularity|roundness|"
r"aspect_ratio|orientation|zernike|moment|hu)(_|$)", MEAN),
(r"(^|_)(centroid|center|centre|coord|bbox|position|_x|_y)(_|$)", MEAN),
)
#: What an unrecognised numeric measurement gets. MEAN rather than SUM: a
#: measurement nobody thought about is more often an intensity or a ratio than
#: an extent, and a wrong mean is a smaller error than a wrong total.
DEFAULT_AGGREGATION = MEAN
#: Non-numeric columns take the first value: text does not add up.
TEXT_AGGREGATION = FIRST
#: What to do with objects that have no children.
NA_POLICIES: Tuple[str, ...] = ("keep", "zero", "drop")
#: Identity columns a merge joins on, and the child's link to its parent.
IDENTITY = ("plateID", "rowID", "columnID", "fieldID")
PARENT_LINK = "cell_id"
PNG_TABLE = "png_list"
OBJECT_COLUMN = "object_label"
#: The tables that can be merged, and the default primary.
OBJECT_TABLES: Tuple[str, ...] = (
"cell", "nucleus", "pathogen", "cytoplasm", *ORGANELLE_ROLES,
)
DEFAULT_PRIMARY = "cell"
[docs]
class MergeError(ValueError):
"""A merge that cannot be done, and why."""
@dataclass(frozen=True)
[docs]
class MergePolicy:
"""How a merge is performed. Every field is a user-facing setting.
:param primary: the object everything is rolled up onto. Decides what a
row of the merged table MEANS.
:param na: what happens to a primary object with no children --
``keep`` leaves NaN, ``zero`` fills with 0, ``drop`` removes the row.
Not interchangeable: a cell with no pathogens genuinely has a pathogen
COUNT of zero, and genuinely has no pathogen mean intensity at all.
:param overrides: column -> aggregation, beating the rules. The rules are
right most of the time, and a default that is right most of the time
is a wrong answer nobody can find the rest of it.
:param consolidate_on_cell: whether many-per-cell child tables restrict
output to cells that contributed a child.
:param keep_uninfected: preserve cells without pathogens or organelles as
the uninfected control population even while consolidating.
"""
primary: str = DEFAULT_PRIMARY
na: str = "keep"
overrides: Mapping[str, str] = None
consolidate_on_cell: bool = True
keep_uninfected: bool = True
[docs]
def __post_init__(self) -> None:
"""Validate the NA policy and retain a private copy of overrides."""
if self.na not in NA_POLICIES:
raise MergeError(
f"na={self.na!r} is not one of {list(NA_POLICIES)}")
object.__setattr__(self, "overrides", dict(self.overrides or {}))
[docs]
def how_for(self, table: str) -> str:
"""Whether ``table`` keeps cells it contributed no rows for.
:param table: child object table whose join mode is requested.
A cell has exactly one cytoplasm and may have multiple nuclei,
pathogens, or organelles. The
many-per-cell tables are rolled up to one row per cell first (see
:func:`roll_up`), and ``consolidate_on_cell`` decides what happens to
a cell the roll-up found nothing for:
consolidate_on_cell=True the analysis is about cells that HAVE
the child, so the join is inner
consolidate_on_cell=False keep the cell and leave the child's
columns NA
Pathogen and organelle tables follow ``keep_uninfected`` so the
default retains uninfected control cells. Set it to ``False`` to
restrict the merged table to infected cells.
A one-row-per-cell table keeps whatever :data:`object_roles.JOIN_HOW`
declares, because there is no consolidation to decide about.
"""
from .object_roles import join_how
name = str(table).strip().lower()
if not self.consolidate_on_cell and not is_one_row_per_cell(name):
return "left"
return join_how(name, keep_uninfected=self.keep_uninfected)
[docs]
def aggregation_for(column: str, *, numeric: bool = True,
overrides: Optional[Mapping[str, str]] = None) -> str:
"""How ``column`` combines when several children roll up into one parent.
:param column: measurement column name matched against the ordered
:data:`AGGREGATION_RULES`.
:param numeric: text columns take the first value whatever their name.
:param overrides: explicit choices, which always win.
:returns: one of :data:`AGGREGATIONS`.
"""
if overrides and column in overrides:
chosen = str(overrides[column])
if chosen not in EXPLICIT_AGGREGATIONS:
raise MergeError(
f"{chosen!r} is not an aggregation; choose from "
f"{list(EXPLICIT_AGGREGATIONS)}")
return chosen
if not numeric:
return TEXT_AGGREGATION
name = str(column).lower()
for pattern, how in AGGREGATION_RULES:
if re.search(pattern, name):
return how
return DEFAULT_AGGREGATION
[docs]
def aggregation_plan(frame: pd.DataFrame, *,
overrides: Optional[Mapping[str, str]] = None,
skip: Sequence[str] = ()) -> Dict[str, str]:
"""The aggregation chosen for every column -- what the user gets shown.
:param frame: child-object table whose columns will be rolled up.
Returned rather than applied silently so the settings panel can display
it and the user can override any of it.
"""
plan: Dict[str, str] = {}
for column in frame.columns:
if column in skip:
continue
numeric = pd.api.types.is_numeric_dtype(frame[column])
plan[column] = aggregation_for(column, numeric=numeric,
overrides=overrides)
return plan
def _connect(db_path: str) -> sqlite3.Connection:
"""Open a measurement database through SQLite's read-only URI."""
return sqlite3.connect(Path(db_path).resolve().as_uri() + "?mode=ro", uri=True, timeout=30)
[docs]
def table_names(db_path: str) -> Tuple[str, ...]:
"""Every table in the database, excluding SQLite's own internals.
:param db_path: path to the SQLite database.
:returns: the table names, sorted.
"""
with _connect(db_path) as db:
rows = db.execute(
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()
return tuple(str(r[0]) for r in rows)
[docs]
def mergeable_tables(db_path: str) -> Tuple[str, ...]:
"""The object tables in this database, in preference order.
:param db_path: path to the SQLite database to inspect.
"""
present = set(table_names(db_path))
return tuple([t for t in OBJECT_TABLES if t in present]
+ ([PNG_TABLE] if PNG_TABLE in present else []))
def _read(db_path: str, table: str) -> pd.DataFrame:
"""Read every row and column from one quoted database table.
Column names are kept as stored, because merge definitions name the
columns the database schema reports.
"""
from .tabular import _read_query
with _connect(db_path) as db:
return _read_query(db, 'SELECT * FROM "' + table.replace('"', '""') + '"',
canonicalise=False, report=None)
def _keys_in(frame: pd.DataFrame) -> List[str]:
"""Return canonical identity columns present, in identity order."""
return [c for c in (*IDENTITY, "timeID", "time_id")
if c in frame.columns]
[docs]
def object_keys(values: pd.Series) -> pd.Series:
"""Object identifiers as integers, whatever spelling they arrived in.
:param values: object-label series to coerce to nullable integers.
The object key is an integer in every object table and TEXT in
``png_list`` -- ``'o5'`` -- so merging the two raised
you are trying to merge on int64 and object columns for key
object_label
which names the dtypes and not the tables, and stopped the whole merge.
The ``'o5'`` form is translated by the one function that already knows
every way it goes wrong (``'omulti'``, ``'onone'``, ``'error'``, NULL);
plain numeric text is converted directly. Anything left becomes NA, so
those rows do not match rather than causing the merge to fail.
"""
if pd.api.types.is_numeric_dtype(values):
return pd.to_numeric(values, errors="coerce").astype("Int64")
text = values.astype("string")
if text.str.match(r"^[a-zA-Z]", na=False).any():
from .utils import object_label_from_png_id
return pd.Series(object_label_from_png_id(values),
index=values.index).astype("Int64")
return pd.to_numeric(text, errors="coerce").astype("Int64")
def _align_keys(left: pd.DataFrame, right: pd.DataFrame,
keys: Sequence[str]) -> None:
"""Make both sides of a merge agree on the TYPE of every key.
In place, on copies the caller owns. Identity columns are compared as
text because a plate called ``1`` is read as an integer from one table and
a string from another depending on what else is in the column -- the same
class of failure as the object key, and just as fatal to a merge.
"""
for key in keys:
if key not in left.columns or key not in right.columns:
continue
if key == OBJECT_COLUMN:
left[key] = object_keys(left[key])
right[key] = object_keys(right[key])
elif not (pd.api.types.is_numeric_dtype(left[key])
and pd.api.types.is_numeric_dtype(right[key])):
left[key] = left[key].astype("string")
right[key] = right[key].astype("string")
[docs]
def aggregation_overrides(policy: MergePolicy, table: str) -> Dict[str, str]:
"""Resolve global rules and table-qualified per-column overrides.
:param policy: Shared merge policy.
:param table: Source table whose aggregation rules are requested.
:returns: Unqualified column-to-method mapping for this table.
"""
overrides = {c: value for c, value in policy.overrides.items() if "." not in c}
overrides.update({c[len(table) + 1:]: value for c, value in policy.overrides.items()
if c.startswith(table + ".")})
return overrides
[docs]
def roll_up(child: pd.DataFrame, keys: Sequence[str], *,
name: str, policy: MergePolicy) -> pd.DataFrame:
"""Aggregate ``child`` onto its parent, one rule per column.
:param child: child-object rows to group and aggregate.
:param keys: the parent's identity in the child -- the identity columns
plus the parent link.
:param name: the child table's name, used to prefix its columns.
:param policy: merge policy supplying per-column aggregation overrides.
:returns: one row per parent, columns prefixed with ``name``.
:raises MergeError: the child has none of the keys.
An unmeasured group is missing, never a measured zero. Regression and both
interactive merge consumers use this same rule.
"""
missing = [k for k in keys if k not in child.columns]
if missing:
raise MergeError(
f"{name} has no {', '.join(missing)}, so its rows cannot be "
f"matched to a parent; re-run Measure with the parent mask set")
overrides = aggregation_overrides(policy, name)
plan = aggregation_plan(child, overrides=overrides, skip=keys)
grouped = child.groupby(list(keys), dropna=False)
out = grouped.agg(plan)
summed = [column for column, how in plan.items() if how == SUM]
if summed:
out[summed] = grouped[summed].sum(min_count=1)
for column, how in plan.items():
if how in ("any", "all"):
out[column] = out[column].astype("boolean").mask(grouped[column].count().eq(0))
count_column = "count"
while count_column in out or f"{name}_{count_column}" in out:
count_column = "source_" + count_column
measured_column = "measured"
while measured_column in out or f"{name}_{measured_column}" in out:
measured_column = "source_" + measured_column
out[count_column] = grouped.size()
measured_columns = [c for c in plan
if plan[c] != FIRST and c in child.columns]
if measured_columns:
non_null = grouped[measured_columns].count()
out[measured_column] = non_null.min(axis=1)
short = {c: int((non_null[c] < out[count_column]).sum())
for c in measured_columns
if (non_null[c] < out[count_column]).any()}
if short:
worst = sorted(short.items(), key=lambda kv: -kv[1])[:5]
LOG.info(
"%s roll-up: %d column(s) had missing values that pandas "
"skips silently; worst affected %s. `%s` carries "
"the smallest contributing count per parent, and is less "
"than `%s` wherever this happened.",
name, len(short),
", ".join(f"{c} ({n} parents)" for c, n in worst),
f"{name}_{measured_column}", f"{name}_{count_column}")
else:
out[measured_column] = out[count_column]
out = out.reset_index()
renamed = {c: f"{name}_{c}" for c in out.columns
if c not in keys and not str(c).startswith(f"{name}_")}
return out.rename(columns=renamed)
#: What happens when the same column arrives from two tables carrying
#: DIFFERENT values for the same object. ``warn`` prints and keeps the
#: left-hand one; ``raise`` stops the analysis.
CONFLICT_POLICIES: Tuple[str, ...] = ("warn", "raise")
#: Columns that name WHICH object a row is, rather than measuring it. Both
#: tables read them off the same image, so they must match -- and when they
#: do not, the two tables are describing different objects under one
#: identity. Everything else is a measurement: cell ``area`` and cytoplasm
#: ``area`` are SUPPOSED to differ, and reporting that as a conflict would
#: bury the real ones.
MUST_AGREE: Tuple[str, ...] = IDENTITY + (
"prc", "prcf", "prcfo", "cell_id", "timeID", "time_id",
"plate_name", "row_name", "column_name", "field_name",
)
#: Columns that must agree only between tables sharing one LABEL SPACE.
#:
#: `object_label` was in MUST_AGREE and should not have been. A cell's
#: object_label is its label in the CELL mask and a pathogen's is its label
#: in the PATHOGEN mask -- two separate labellings of two separate objects,
#: with no reason on earth to coincide. Joining cell to pathogen therefore
#: warned about nearly every row of every healthy screen:
#:
#: 'object_label': 60095 of 60816 objects disagree between cell and
#: pathogen (e.g. prcfo 0, 1, 2, 3, 4)
#:
#: which says "a defect in the data no analysis should quietly average
#: over" about data that is exactly right. A warning that fires on the
#: normal case teaches its reader to ignore it, and the real conflicts it
#: was built for go with it.
#:
#: Cytoplasm is the exception and the reason the column is not simply
#: dropped from the check: it is the cell minus its nucleus, carries the
#: CELL's label, and a disagreement there is a genuine mismatch.
SAME_LABEL_SPACE: Dict[str, Tuple[str, ...]] = {
"object_label": ("cell", "cytoplasm"),
}
def _shares_a_label_space(column: str, left_name: str, right_name: str) -> bool:
"""Whether ``column`` has to agree between these two tables in particular.
:data:`MUST_AGREE` is a flat list and cannot express "these two columns
are the same fact only when the tables are related". `object_label` is
that case -- see :data:`SAME_LABEL_SPACE`.
:returns: True when both table names are in the column's label space.
"""
space = SAME_LABEL_SPACE.get(str(column))
if not space:
return False
return all(any(name in str(side) for name in space)
for side in (left_name, right_name))
[docs]
class ColumnConflict(MergeError):
"""One column, two tables, two different values for the same object."""
def _columns_agree(left: pd.Series, right: pd.Series) -> pd.Series:
"""Row-wise equality where a missing value cannot disagree with anything.
``NaN != NaN`` is right for arithmetic and wrong here: a column absent
from both tables for a given object is not a disagreement about it.
EITHER side missing, not only both. An UNINFECTED CELL is a cell kept on
purpose -- ``keep_uninfected=True`` -- that has no pathogen row, so every
column from the pathogen table is absent for it. Comparing a present
plateID against that absence counted as a conflict, and plate1 reported
'plateID': 172 of 553 objects disagree between cell and pathogen
where 172 is exactly the number of cells with no pathogen in them. The
data is right, the cells were kept deliberately, and the identity columns
were being compared against nothing at all.
"""
both_missing = left.isna() | right.isna()
if (pd.api.types.is_numeric_dtype(left)
and pd.api.types.is_numeric_dtype(right)):
same = pd.Series(
np.isclose(pd.to_numeric(left, errors="coerce"),
pd.to_numeric(right, errors="coerce"),
rtol=1e-9, atol=1e-12, equal_nan=True),
index=left.index)
else:
same = left.astype(object).eq(right.astype(object))
return same | both_missing
[docs]
def reconcile_duplicates(frame: pd.DataFrame, suffix: str, *,
key: str = "prcfo",
left_name: str = "the primary table",
right_name: str = "the joined table",
on_conflict: str = "warn") -> pd.DataFrame:
"""Collapse ``col``/``col+suffix`` pairs that agree; report those that do not.
Joining two measurement tables gives every shared column twice --
``plateID`` and ``plateID_cytoplasm`` hold the same plate written by two
stages of the same run. Carrying both doubles the width of the frame and
invites a downstream reader to pick the wrong one.
So the pair is COMPARED, object by object, rather than assumed: identical
columns collapse to one, and a column that disagrees means the two tables
describe different objects under the same identity, which is a defect in
the data no analysis should quietly average over.
:param frame: the merged frame, modified only by dropping columns.
:param suffix: what the merge appended to the right-hand duplicates.
:param key: the identity the comparison is reported against.
:param on_conflict: ``warn`` keeps the left-hand column and prints;
``raise`` stops with :class:`ColumnConflict`.
:returns: the frame with agreeing duplicates dropped.
:raises ColumnConflict: a pair disagrees and ``on_conflict='raise'``.
"""
if on_conflict not in CONFLICT_POLICIES:
raise MergeError(
f"on_conflict={on_conflict!r} is not one of "
f"{list(CONFLICT_POLICIES)}")
if not suffix:
return frame
drop, conflicts = [], []
for right_col in [c for c in frame.columns if str(c).endswith(suffix)]:
left_col = str(right_col)[: -len(suffix)]
if left_col not in frame.columns:
continue
agree = _columns_agree(frame[left_col], frame[right_col])
if bool(agree.all()):
drop.append(right_col)
continue
if left_col not in MUST_AGREE and not _shares_a_label_space(
left_col, left_name, right_name):
continue
disagreeing = frame.index[~agree]
where = (frame.loc[disagreeing, key].astype(str).tolist()[:5]
if key in frame.columns else
[str(i) for i in disagreeing[:5]])
conflicts.append(
f"{left_col!r}: {int((~agree).sum())} of {len(agree)} objects "
f"disagree between {left_name} and {right_name}"
+ (f" (e.g. {key} " + ", ".join(where) + ")" if where else ""))
if conflicts:
detail = ("the same column arrived from two tables with different "
"values for the same object:\n "
+ "\n ".join(conflicts))
if on_conflict == "raise":
raise ColumnConflict(detail)
LOG.warning(detail)
print(f"WARNING: {detail}")
return frame.drop(columns=drop) if drop else frame
[docs]
def merge_tables(db_path: str, tables: Sequence[str], *,
policy: Optional[MergePolicy] = None) -> pd.DataFrame:
"""One table with every chosen object's measurements on it.
This is what makes "a cell measurement on one axis, nuclear on another and
pathogen on a third" possible: each table's columns arrive prefixed with
the object they measure, so they can be told apart and picked separately.
:param db_path: path to the SQLite measurements database.
:param tables: which object tables to include. The primary must be one of
them, and is added if it is not.
:param policy: how to aggregate and what to do with childless parents.
:returns: one row per primary object.
:raises MergeError: the primary table is not in the database, or a child
cannot be linked to it.
"""
policy = policy or MergePolicy()
available = set(table_names(db_path))
if policy.primary not in available:
raise MergeError(
f"the database has no {policy.primary!r} table to merge onto; it "
f"has {', '.join(sorted(available & set(OBJECT_TABLES))) or 'none'}")
wanted = [t for t in dict.fromkeys([policy.primary, *tables])
if t in available]
if PNG_TABLE in tables and PNG_TABLE not in wanted and PNG_TABLE in available:
wanted.append(PNG_TABLE)
base = _read(db_path, policy.primary)
keys = _keys_in(base)
if OBJECT_COLUMN not in base.columns:
raise MergeError(
f"{policy.primary} has no {OBJECT_COLUMN}, so nothing can be "
f"merged onto it")
merged = base.rename(
columns={c: f"{policy.primary}_{c}" for c in base.columns
if c not in keys + [OBJECT_COLUMN]})
for table in wanted:
if table == policy.primary:
continue
child = _read(db_path, table)
if table == PNG_TABLE:
merged = _merge_crops(merged, child, keys)
continue
anchor = anchor_column(table) if table in ANCHOR_COLUMN else PARENT_LINK
if anchor not in child.columns:
LOG.info("%s carries no %s, so it cannot be joined onto %s; "
"leaving it out", table, anchor, policy.primary)
continue
if is_one_row_per_cell(table):
skip = set(_keys_in(child)) | {anchor, "prcf", "prcfo"}
rolled = child.rename(
columns={c: (c if c.startswith(f"{table}_") else f"{table}_{c}")
for c in child.columns if c not in skip})
else:
child_keys = _keys_in(child) + [anchor]
rolled = roll_up(child, child_keys, name=table, policy=policy)
rolled = rolled.rename(columns={anchor: OBJECT_COLUMN})
on = [c for c in keys + [OBJECT_COLUMN] if c in rolled.columns]
if OBJECT_COLUMN not in on:
LOG.info("%s cannot be joined to %s on an object key", table,
policy.primary)
continue
_align_keys(merged, rolled, on)
how = policy.how_for(table)
before = len(merged)
merged = merged.merge(rolled, on=on, how=how, validate="many_to_one")
if how == "inner" and len(merged) < before:
LOG.info(
"%s joined %s: %d of %d %s objects had no %s row and were "
"removed (consolidate_on_cell=%s, keep_uninfected=%s)",
how, table, before - len(merged), before, policy.primary,
table, policy.consolidate_on_cell, policy.keep_uninfected)
return _apply_na_policy(merged, policy)
def _merge_crops(merged: pd.DataFrame, png: pd.DataFrame,
keys: Sequence[str]) -> pd.DataFrame:
"""Attach crop paths from ``png_list``.
Not aggregated: a crop is not a measurement, and png_list is one row per
crop rather than per object. Its object id may be under any of the
crop-mode columns, so the first one present is used -- a database measured
in more than one crop mode has several, and the others belong to different
objects entirely.
"""
from .utils import PNG_OBJECT_ID_COLUMNS
id_column = next((c for c in PNG_OBJECT_ID_COLUMNS.values()
if c in png.columns), None)
if id_column is None:
LOG.info("%s carries no object id column; no crop paths merged",
PNG_TABLE)
return merged
path_column = next((c for c in ("png_path", "path", "file_path")
if c in png.columns), None)
side = png[[c for c in keys if c in png.columns]].copy()
side[OBJECT_COLUMN] = object_keys(png[id_column])
if path_column:
side[f"{PNG_TABLE}_path"] = png[path_column]
side = side.dropna(subset=[OBJECT_COLUMN])
on = [c for c in list(keys) + [OBJECT_COLUMN] if c in side.columns]
side = side.drop_duplicates(subset=on)
_align_keys(merged, side, on)
return merged.merge(side, on=on, how="left")
def _apply_na_policy(frame: pd.DataFrame, policy: MergePolicy) -> pd.DataFrame:
"""What happens to a primary object with no children.
``zero`` fills only the COUNTS by default reasoning: a cell with no
pathogens has a pathogen count of zero, but it does not have a pathogen
mean intensity of zero -- it has none, and zero would be a measurement
that was never made. Callers who want the blunter behaviour choose it
explicitly.
"""
counts = [c for c in frame.columns if c.endswith("_count")]
if policy.na == "keep":
for column in counts:
frame[column] = frame[column].fillna(0)
return frame
if policy.na == "zero":
return frame.fillna({c: 0 for c in frame.columns
if pd.api.types.is_numeric_dtype(frame[c])})
if policy.na == "drop":
if not counts:
return frame.reset_index(drop=True)
return frame.dropna(subset=counts, how="any").reset_index(drop=True)
return frame
#: Reductions offered in xD mode.
REDUCTIONS: Tuple[str, ...] = ("pca", "umap", "tsne")
[docs]
class ReductionError(ValueError):
"""A reduction that cannot be computed, and why."""
[docs]
def reduce_dimensions(frame: pd.DataFrame, columns: Sequence[str], *,
method: str = "pca", components: int = 2,
scale: bool = True, min_coverage: float = 0.5,
seed: int = 0, n_neighbors: int = 15,
min_dist: float = 0.1,
perplexity: float = 30.0) -> pd.DataFrame:
"""Reduce many measurements to a few, for gating in xD.
Gating in more dimensions than can be drawn means drawing something else:
a projection. The components come back as ORDINARY COLUMNS (``PC1``,
``PC2``, ...), so every existing gate tool works on them unchanged -- a
gate on PC1 vs PC2 is the same kind of object as a gate on area vs
intensity, and saves, re-applies and exports identically.
:param frame: object-by-measurement table to project. The returned frame is
reindexed to this table's complete index.
:param columns: the measurements to reduce. At least two.
:param method: ``pca`` always available; umap and t-SNE if installed.
:param scale: standardise first. Without it a measurement whose numbers
are larger dominates every component regardless of what it means.
:param min_coverage: a column with fewer than this fraction of real values
is left out. What remains is median-filled rather than row-dropped --
see the comment in the body, which is the difference between xD
working on a real table and returning nothing at all.
:param n_neighbors: UMAP only. How much of the data each point is placed
against: small values keep local structure and fragment the map,
large ones preserve the global shape and merge populations. Ignored
by PCA and t-SNE, which have no such parameter -- hence the greying
in the xD tab rather than a control that silently does nothing.
:param min_dist: UMAP only. How tightly points may pack. Ignored elsewhere.
:param perplexity: t-SNE only, and CLAMPED to ``(n - 1) / 3``: sklearn
raises outright when it exceeds the sample size, which would turn a
legitimate setting into a failed projection on a small selection.
:returns: a frame of components, indexed like ``frame``.
:raises ReductionError: too few columns, too few rows, nothing numeric, or
a method whose package is not installed.
"""
if method not in REDUCTIONS:
raise ReductionError(
f"{method!r} is not one of {list(REDUCTIONS)}")
chosen = [c for c in columns if c in frame.columns]
if len(chosen) < 2:
raise ReductionError(
"reducing needs at least two measurements; pick more columns")
data = frame[chosen].apply(pd.to_numeric, errors="coerce")
coverage = data.notna().mean()
keep = [c for c in chosen if coverage.get(c, 0.0) >= min_coverage]
dropped = [c for c in chosen if c not in keep]
if dropped:
LOG.info("%d column(s) are under %.0f%% complete and were left out of "
"the projection: %s", len(dropped), min_coverage * 100,
", ".join(dropped[:6]) + ("…" if len(dropped) > 6 else ""))
if len(keep) < 2:
raise ReductionError(
f"only {len(keep)} of {len(chosen)} measurement(s) are at least "
f"{min_coverage:.0%} complete, and a projection needs two; lower "
f"the coverage requirement or pick fuller columns")
data = data[keep]
usable = data.fillna(data.median(numeric_only=True))
usable = usable.dropna(axis=1, how="any")
if usable.shape[1] < 2:
raise ReductionError(
"no two measurements have enough values in common to project")
usable = usable.loc[data.notna().any(axis=1)]
if len(usable) < 3:
raise ReductionError(
f"only {len(usable)} object(s) have any of these measurements; "
f"there is nothing to project")
chosen = list(usable.columns)
components = max(2, min(int(components), len(chosen), len(usable)))
values = usable.to_numpy(dtype=float)
if scale:
centre = values.mean(axis=0)
spread = values.std(axis=0)
spread[spread == 0] = 1.0
values = (values - centre) / spread
if method == "pca":
from sklearn.decomposition import PCA
model = PCA(n_components=components, random_state=seed)
reduced = model.fit_transform(values)
names = [f"PC{i + 1}" for i in range(reduced.shape[1])]
out = pd.DataFrame(reduced, index=usable.index, columns=names)
out.attrs["explained_variance"] = list(
getattr(model, "explained_variance_ratio_", []))
elif method == "umap":
try:
from .utils import umap
_ = umap.UMAP
except Exception as exc:
raise ReductionError(
"UMAP is not installed in this environment; PCA is always "
"available") from exc
reduced = umap.UMAP(n_components=components,
n_neighbors=max(2, min(int(n_neighbors),
len(values) - 1)),
min_dist=float(min_dist),
random_state=seed).fit_transform(values)
out = pd.DataFrame(reduced, index=usable.index,
columns=[f"UMAP{i + 1}" for i in range(components)])
else:
from sklearn.manifold import TSNE
bounded = max(5.0, min(float(perplexity), (len(values) - 1) / 3.0))
reduced = TSNE(n_components=min(components, 3), perplexity=bounded,
random_state=seed).fit_transform(values)
out = pd.DataFrame(reduced, index=usable.index,
columns=[f"tSNE{i + 1}"
for i in range(reduced.shape[1])])
return out.reindex(frame.index)
[docs]
def group_variance_share(frame: pd.DataFrame,
groups: Dict[str, Sequence[str]], *,
scale: bool = True,
min_coverage: float = 0.5) -> pd.DataFrame:
"""Calculate each column group's share of variance in the input matrix.
The matrix is prepared with the same missing-value and scaling procedure
used by :func:`reduce_dimensions`. The result therefore characterizes the
inputs supplied to PCA, UMAP, t-SNE, and other reducers without requiring
method-specific loadings.
:param frame: Measurement frame containing the candidate feature columns.
:param groups: ``{group_name: columns}``. A column named by two groups is
counted in both; shares may therefore sum to more than one, and this
condition is recorded in ``result.attrs['overlapping']``.
:param scale: Whether to standardize features before calculating variance,
matching the corresponding reducer option.
:param min_coverage: Minimum non-missing fraction required for a feature to
enter the prepared matrix.
:returns: DataFrame indexed by group with ``share`` and ``columns`` fields,
sorted by decreasing share.
"""
prepared, used = _prepared_matrix(frame, [c for cols in groups.values()
for c in cols],
scale=scale, min_coverage=min_coverage)
if prepared.empty:
return pd.DataFrame(columns=["share", "columns"])
variance = prepared.var(axis=0)
rows = {}
for name, columns in groups.items():
present = [c for c in columns if c in used]
rows[name] = (float(variance[present].sum()) if present else 0.0,
len(present))
total = sum(value for value, _count in rows.values())
out = pd.DataFrame(
[{"group": name, "share": (value / total if total else 0.0),
"columns": count}
for name, (value, count) in rows.items()]).set_index("group")
out.attrs["overlapping"] = bool(
sum(len([c for c in cols if c in used]) for cols in groups.values())
> len(used))
return out.sort_values("share", ascending=False)
[docs]
def missingness_leak(components: pd.DataFrame, frame: pd.DataFrame,
columns: Sequence[str], *,
min_objects: int = 30) -> pd.DataFrame:
"""Does "was this object measured" predict where it landed?
For each column, the objects that HAVE a value and the objects that do
not are compared by the distance between their centroids in the
projection, expressed in map radii so it is comparable across runs and
across methods.
A gap near 1 means the projection has separated the two groups about as
far as the map is wide -- on the fact of measurement, not on a
measurement. In spaCR that is usually infected against uninfected, and it
is exactly the kind of split a user would otherwise write up.
:param components: the reducer's output, indexed like ``frame``.
:param frame: original measurement table used to determine which component
rows had or lacked each input measurement.
:param columns: the columns that went into the projection.
:param min_objects: skip a column unless both sides have at least this
many objects. A gap computed from four objects is noise, and
reporting it would bury the real ones.
:returns: one row per checked column, worst ``severity`` first. Empty --
WITH ITS COLUMNS -- when nothing was checkable, so a caller can sort
it without a KeyError.
TWO ARTEFACTS, NOT ONE, and this is where spaCR differs from the tool the
idea came from. Which one appears depends on how the gap was filled:
``centroid_gap``
the missing objects sit SOMEWHERE ELSE. Near 1 means the projection
has moved them about as far as the map is wide.
``dispersion_ratio``
the missing objects COLLAPSE. ``reduce_dimensions`` fills with the
column median, so every uninfected cell gets the SAME value on every
pathogen column and they land on one point. Near 0 means they have
no spread of their own.
Measured on a synthetic infected/uninfected table with twelve pathogen
columns, the median fill produced a centroid gap of 0.06 -- almost
nothing -- and a dispersion ratio of 0.11. THE CENTROID STATISTIC ALONE
WOULD HAVE MISSED IT, because a median fill puts the missing objects in
the middle of the present ones rather than away from them. Both are
reported, and ``severity`` is whichever is worse.
"""
empty = pd.DataFrame(columns=["column", "missing_fraction", "centroid_gap",
"dispersion_ratio", "severity"])
axes = components.select_dtypes("number").dropna()
if axes.empty or axes.shape[1] < 2:
return empty
coords = axes.to_numpy(dtype=float)
radius = float(np.sqrt((((coords - coords.mean(axis=0)) ** 2).sum(axis=1)).mean()))
if not radius:
return empty
rows = []
for column in columns:
if column not in frame.columns:
continue
missing = frame.loc[axes.index, column].isna().to_numpy()
if missing.sum() < min_objects or (~missing).sum() < min_objects:
continue
gap = float(np.linalg.norm(
coords[missing].mean(axis=0) - coords[~missing].mean(axis=0)))
absent, present = _spread(coords[missing]), _spread(coords[~missing])
ratio = absent / present if present else 1.0
rows.append({"column": column,
"missing_fraction": float(missing.mean()),
"centroid_gap": gap / radius,
"dispersion_ratio": ratio,
"severity": max(gap / radius, 1.0 - min(ratio, 1.0))})
if not rows:
return empty
return pd.DataFrame(rows).sort_values("severity", ascending=False)
def _spread(points) -> float:
"""Root-mean-square distance of ``points`` from their own centroid."""
if len(points) == 0:
return 0.0
return float(np.sqrt((((points - points.mean(axis=0)) ** 2)
.sum(axis=1)).mean()))
def _prepared_matrix(frame: pd.DataFrame, columns: Sequence[str], *,
scale: bool, min_coverage: float):
"""The matrix :func:`reduce_dimensions` would build, and its columns.
Kept in step with the reducer on purpose: a diagnostic computed on a
differently-prepared matrix describes a projection nobody ran.
"""
chosen = list(dict.fromkeys(c for c in columns if c in frame.columns))
if len(chosen) < 2:
return pd.DataFrame(), []
data = frame[chosen].apply(pd.to_numeric, errors="coerce")
coverage = data.notna().mean()
keep = [c for c in chosen if coverage.get(c, 0.0) >= min_coverage]
if len(keep) < 2:
return pd.DataFrame(), []
data = data[keep]
usable = data.fillna(data.median(numeric_only=True)).dropna(axis=1, how="any")
usable = usable.loc[data.notna().any(axis=1)]
if usable.shape[1] < 2 or len(usable) < 3:
return pd.DataFrame(), []
if scale:
centre = usable.mean(axis=0)
spread = usable.std(axis=0).replace(0.0, 1.0)
usable = (usable - centre) / spread
return usable, list(usable.columns)