"""Write per-object classification scores back into a spaCR ``measurements.db``.
Two classifiers in this package score *every object of a whole database*:
* the convolutional one --- :func:`spacr.deep_spacr.apply_model_to_tar`, driven
by :func:`spacr.deep_spacr.deep_spacr`; and
* the classical-ML one --- :func:`spacr.ml.generate_ml_scores`.
Both produce one row per crop. The Annotate app, the active-learning queue,
``generate_image_umap``, the plate heatmaps and every GUI table read the
``png_list`` table of ``measurements.db``, so the scores are written there, on
the row they belong to.
Why this is its own module
--------------------------
The merge is not in :mod:`spacr.deep_spacr`, because importing ``deep_spacr``
pulls in torch and torchvision, which the classical-ML path does not use. It is
not in :mod:`spacr.utils`, which owns the database *write* helpers
(``_append_to_measurements_db``, ``rename_columns_in_db``) but is 8000 lines
and imports most of the package. This module is small, imports nothing from
spaCR at module scope except :mod:`spacr.utils` helpers pulled in lazily, and
is what both callers import.
The join key
------------
``prcfo`` --- ``plate_row_column_field_object`` --- is the canonical per-object
identity in this codebase. :func:`spacr.utils.filepaths_to_database` writes it
onto every ``png_list`` row, :func:`spacr.io._read_and_merge_data` indexes the
merged feature frame by it, and :func:`spacr.ml.generate_ml_scores` already
joins annotations onto features with it. Keying the merge on it means the CV
and the ML stage write to *the same row*.
A basename such as ``os.path.basename(png_path)`` is not an identity:
* the tar handed to the model is built with
``arcname=os.path.basename(img_path)`` (:func:`spacr.utils.add_images_to_tar`),
so a run over two source folders whose plates are *both* called ``plate1``
--- which is what happens whenever the plate name comes from the source
folder name, see :mod:`spacr.io` ``_rename_and_organize_image_files`` ---
puts two different crops in the archive under one member name; and
* a plain ``dict`` assignment would let the second of those two crops silently
overwrite the first, scoring one plate with the other plate's predictions.
Changing the key does not prevent this: two crops that share a basename share a
``prcfo`` too. The collision is therefore *detected*. A key that arrives twice
with two different values is recorded as ambiguous, written nowhere, and
counted in the report. A wrong score is worse than a missing one.
Key selection is measured, not assumed. Candidate keys are built for both sides
(``prcfo``, then the full ``png_path``, then ``file_name`` --- never a basename
computed behind the caller's back when a real column exists) and the one that
matches the most rows is used, ties going to the earliest in
:data:`KEY_PRIORITY`. The chosen key and every count are printed, so a merge
that matched 3 of 40000 rows can be told apart from one that matched all of
them.
Atomicity
---------
Python's :mod:`sqlite3` opens an implicit transaction for DML only, so an
``ALTER TABLE`` runs in autocommit and lands immediately --- the same trap
:func:`spacr.utils.rename_columns_in_db` was fixed for. The transaction here is
opened explicitly and rolled back on any error, so an interrupted merge leaves
the table exactly as it was: no half-added column, no half-scored rows.
"""
from __future__ import annotations
import math
import os
import sqlite3
from dataclasses import dataclass
from typing import Dict, List, Mapping, Optional, Sequence, Tuple
import numpy as np
import pandas as pd
__all__ = [
"PNG_TABLE",
"CV_SCORE_COLUMN",
"CV_CLASS_COLUMN",
"ML_SCORE_COLUMN",
"ML_CLASS_COLUMN",
"ANNOTATE_ENCODED_COLUMNS",
"KEY_PRIORITY",
"MergeReport",
"crop_name_metadata",
"migrate_prediction_columns",
"merge_prediction_results",
"merge_cv_predictions",
"merge_ml_predictions",
]
#: Table holding one row per object crop.
PNG_TABLE = "png_list"
#: Positive-class probability from the convolutional classifier. Spelled
#: ``pred`` because that is what it has always been called, and because
#: ``spacr.settings`` (``dependent_variable='pred'``), ``spacr.submodules``,
#: ``spacr.plot`` and ``spacr.active_learning.PRED_COLUMN_CANDIDATES`` all read
#: that name; renaming it would break saved settings files for a tidier column.
CV_SCORE_COLUMN = "pred"
#: Thresholded class from the convolutional classifier
#: (:func:`spacr.utils.process_vision_results`).
CV_CLASS_COLUMN = "cv_predictions"
#: What a per-object classification score is CALLED in a score table, in the
#: order they are tried. Which name it gets depends on which classifier
#: wrote the file, not on anything the reader chose:
#:
#: * ``pred`` -- `process_vision_results` and the deep-learning CV scores;
#: * ``prediction_probability_class_1`` -- `ml_analysis`, i.e. the XGBoost
#: and other scikit-learn fits, whose positive-class probability this is.
#:
#: Kept in ONE place because the two readers were already meant to agree:
#: :func:`merge_ml_predictions` knew the second name and
#: :func:`attach_predictions` did not, so merging an XGBoost score file into
#: a database worked while reading the same file in memory failed with "no
#: 'pred' column" -- about a file that held the score all along.
SCORE_SOURCE_COLUMNS: Tuple[str, ...] = (
CV_SCORE_COLUMN, "prediction_probability_class_1",
)
#: Likewise for the predicted CLASS.
CLASS_SOURCE_COLUMNS: Tuple[str, ...] = (CV_CLASS_COLUMN, "predictions")
[docs]
def first_present(frame, names) -> Optional[str]:
"""The first of ``names`` that ``frame`` actually has, else None."""
found = getattr(frame, "columns", None)
columns = set() if found is None else {str(name) for name in found}
for name in names:
if str(name) in columns:
return str(name)
return None
#: Positive-class probability from the classical-ML classifier. This one is
#: new: the ML stage only ever wrote a class, never its confidence. Namespaced
#: rather than reusing ``pred`` precisely so it cannot collide with the CV
#: score -- that collision is what "CV and ML must coexist" is about.
ML_SCORE_COLUMN = "ml_pred"
#: Predicted class from the classical-ML classifier. Kept as ``predictions``,
#: the spelling :func:`spacr.utils.add_column_to_database` has always written,
#: because ``spacr.settings`` names it in two defaults that read ``png_list``
#: --- ``set_analyze_endodyogeny_defaults``' ``class_column`` and
#: ``set_default_analyze_screen``' ``heatmap_feature``. A tidier
#: ``ml_predictions`` would silently break both for anyone who re-ran the ML
#: stage, and the name it would be tidier than is not ambiguous in practice:
#: the CV stage writes ``cv_predictions``, so the two never share a column.
ML_CLASS_COLUMN = "predictions"
#: ``png_list`` class columns an old spaCR wrote in the Annotate app's label
#: encoding. :func:`spacr.utils.add_column_to_database` replaced every ``0``
#: with a ``2`` before storing, because the Annotate app labels classes 1 and
#: 2 -- so the database disagreed with the ``results.csv`` written by the same
#: run. :func:`migrate_prediction_columns` puts them back.
ANNOTATE_ENCODED_COLUMNS: Tuple[str, ...] = (ML_CLASS_COLUMN,)
#: Candidate join keys, best first. Ties in match count are broken by this
#: order, so ``prcfo`` wins whenever it does as well as the alternatives.
KEY_PRIORITY: Tuple[str, ...] = ("prcfo", "png_path", "file_name")
#: Columns a results frame may carry the crop's name in, best first.
_NAME_COLUMNS: Tuple[str, ...] = ("path", "png_path", "file_name")
#: Per-crop-mode object-id columns :func:`spacr.utils.filepaths_to_database`
#: writes ('o<n>' strings). Used to rebuild ``prcfo`` when a table somehow
#: lacks the column but still carries the metadata it is made of.
#:
#: DERIVED FROM THE ROLES, NOT LISTED BY HAND. This was a hand-written copy of
#: spacr.utils.PNG_OBJECT_ID_COLUMNS and it had drifted: the four organelle
#: roles were absent, so an organelle-mode score table could not rebuild its
#: key and its join matched zero rows -- read as "no per-object score", which
#: is the exact failure the comment in ``_result_keys`` says was fixed for the
#: plainer spellings. PNG_OBJECT_ID_COLUMNS carries a comment about organelle
#: having been missing from IT once, for the same reason.
#:
#: ``spacr.utils`` is imported lazily elsewhere in this module to avoid a
#: cycle, so the roles come from ``spacr.schema`` -- which is where
#: PNG_OBJECT_ID_COLUMNS gets them too.
#:
#: Order is precedence, and the two additions go LAST so no table that
#: resolved before resolves differently now.
from .schema import OBJECT_KEY as _OBJECT_KEY, ORGANELLE_ROLES as _ORGANELLE_ROLES
_OBJECT_ID_COLUMNS: Tuple[str, ...] = (
"cell_id", "nucleus_id", "pathogen_id", "cytoplasm_id", "object",
*(f"{role}_id" for role in _ORGANELLE_ROLES),
_OBJECT_KEY,
)
#: Metadata columns ``prcfo`` is assembled from, in order.
_PRCFO_METADATA: Tuple[str, ...] = ("plateID", "rowID", "columnID", "fieldID")
#: What :func:`spacr.utils._map_wells_png` returns for a name it cannot parse.
_UNPARSED = "error"
#: SQLite's three spellings of the implicit row id, least likely to be shadowed
#: first. **Not paranoia**: ``png_list`` has a column called ``rowID``, SQLite
#: identifiers are case-insensitive, and a table that declares a column named
#: ``rowid`` (in any case) makes the bare name resolve to *that column* rather
#: than to the row id. So on a real measurements database ``SELECT rowid`` from
#: ``png_list`` returns the plate row -- ``'r1'``, ``'r2'`` -- and
#: ``UPDATE ... WHERE rowid = 'r1'`` rewrites **every crop in plate row 1**.
#: The merge this module replaces did exactly that; it went unnoticed because
#: its tests built ``png_list`` by hand without the ``rowID`` column.
_ROWID_ALIASES: Tuple[str, ...] = ("_rowid_", "oid", "rowid")
_MISSING = object()
def _quote(identifier: str) -> str:
"""Return ``identifier`` quoted for SQLite.
Column and table names reach this module from settings files and from
caller keyword arguments, and were previously interpolated into
``ALTER TABLE``/``UPDATE`` raw. A name with a space, a reserved word or a
quote character produced a syntax error at best.
:param identifier: table or column name.
:returns: the name wrapped in double quotes, internal quotes doubled.
:raises ValueError: if ``identifier`` is not a non-empty string.
"""
if not isinstance(identifier, str) or not identifier:
raise ValueError(f"Invalid SQL identifier: {identifier!r}")
return '"' + identifier.replace('"', '""') + '"'
def _rowid_alias(table_columns: Sequence[str]) -> str:
"""Return a spelling of the implicit row id this table does not shadow.
:param table_columns: the table's declared column names.
:returns: ``'_rowid_'``, ``'oid'`` or ``'rowid'``.
:raises ValueError: if the table declares all three, leaving no way to
address a row by identity.
"""
taken = {str(name).lower() for name in table_columns}
for alias in _ROWID_ALIASES:
if alias not in taken:
return alias
raise ValueError(
"Table declares columns named rowid, oid and _rowid_, so its rows "
"cannot be addressed by row id; rename one of them before merging.")
def _sql_value(value, sql_type: str):
"""Coerce one DataFrame cell to something SQLite can store.
NaN becomes NULL rather than the float ``nan`` SQLite would otherwise
store as a REAL, and numpy scalars are unwrapped so the driver does not
have to guess.
A value the declared type cannot hold becomes NULL rather than failing the
whole merge, and a value of no type SQLite can bind at all is stored as its
text -- one odd cell in a results frame must not cost 40000 scored rows.
:param value: the cell.
:param sql_type: ``'REAL'``, ``'INTEGER'``, or anything else (stored as a
bindable scalar).
:returns: ``None``, or something :mod:`sqlite3` can bind.
"""
if value is None:
return None
if isinstance(value, np.generic):
value = value.item()
if isinstance(value, float) and math.isnan(value):
return None
try:
if pd.isna(value):
return None
except (TypeError, ValueError):
pass
if sql_type == "INTEGER":
try:
return int(value)
except (TypeError, ValueError):
return None
if sql_type == "REAL":
try:
return float(value)
except (TypeError, ValueError):
return None
if isinstance(value, (int, float, str, bytes)):
return value
return str(value)
def _clean_prcfo(value) -> Optional[str]:
"""A ``prcfo`` key with the plate id in the form spaCR keys on.
THE PLATE IS THE HALF THAT DISAGREES. A screen written by an older run
stamps its plate `pplate1` while everything computed since stamps it
`plate1`, and `schema.canonical_plate_id` is the one rule that collapses
the doubled prefix. It is applied on read for the columns in
`PLATE_BEARING_COLUMNS` -- which includes `prcfo` -- but only by two
callers, and neither is on this path.
So a classifier's scores keyed `plate1_r10_c11_f10_o101` met a png_list
keyed `pplate1_r8_c19_f11_o84` and NOTHING matched: an ML run over
60,816 real cells fitted, scored, explained and plotted, then wrote zero
scores back and said the results "probably come from a different
experiment", about the same database it had just read.
Applied to the prcfo key only. A png_path or a file_name is not a plate
id and must not have its first two characters rewritten.
"""
text = _clean_key(value)
if text is None:
return None
from . import schema
return schema.canonical_plate_id(text)
def _clean_key(value) -> Optional[str]:
"""Return ``value`` as a usable key string, or ``None`` when it is not one."""
if value is None:
return None
try:
if pd.isna(value):
return None
except (TypeError, ValueError):
pass
text = str(value).strip()
if not text or text == _UNPARSED:
return None
return text
def _values_equal(left: Sequence, right: Sequence) -> bool:
"""Whether two rows of already-coerced values are the same.
A plain comparison is enough because every element has been through
:func:`_sql_value`, which turns NaN into ``None`` -- so the one case that
would need special handling (``nan != nan``) cannot reach here.
"""
return tuple(left) == tuple(right)
def _name_column(frame: pd.DataFrame) -> Optional[str]:
"""Return the column holding the crop's name/path, or ``None``."""
for name in _NAME_COLUMNS:
if name in frame.columns:
return name
return None
def _prcfo_from_metadata(frame: pd.DataFrame) -> Optional[pd.Series]:
"""Rebuild ``prcfo`` from the metadata columns ``png_list`` carries.
``filepaths_to_database`` writes ``plateID``/``rowID``/``columnID``/
``fieldID`` (plus ``timeID`` for a timelapse) and one ``<mode>_id`` column
holding the ``'o<n>'`` object id, which is exactly what ``prcfo`` is joined
from. A table that has the metadata but not the key can therefore still be
merged into.
:param frame: rows of the target table.
:returns: a Series of keys, or ``None`` if the metadata is not all there.
"""
from .utils import _time_column
if not all(col in frame.columns for col in _PRCFO_METADATA):
return None
object_col = None
for candidate in _OBJECT_ID_COLUMNS:
if candidate in frame.columns:
object_col = candidate
break
if object_col is None:
return None
parts: List[str] = list(_PRCFO_METADATA)
time_col = _time_column(frame.columns)
if time_col is not None:
parts.append(time_col)
parts.append(object_col)
pieces = []
valid = None
for col in parts:
text = frame[col].astype("object").map(_clean_key)
present = text.notna()
valid = present if valid is None else (valid & present)
pieces.append(text.fillna(""))
key = pieces[0].astype(str)
for piece in pieces[1:]:
key = key + "_" + piece.astype(str)
return pd.Series(
(
_clean_prcfo(value) if bool(is_valid) else None
for value, is_valid in zip(key, valid)
),
index=frame.index,
dtype=object,
)
def _prcfo_from_names(names: pd.Series, timelapse: bool) -> pd.Series:
"""Derive just the ``prcfo`` key from crop file names."""
return crop_name_metadata(names, timelapse=timelapse)["prcfo"]
def _db_keys(kind: str, frame: pd.DataFrame) -> Optional[pd.Series]:
"""Build the ``kind`` key for rows already in the database."""
if kind == "prcfo":
if "prcfo" in frame.columns:
return frame["prcfo"].map(_clean_prcfo)
return _prcfo_from_metadata(frame)
if kind == "png_path":
if "png_path" in frame.columns:
return frame["png_path"].map(_clean_key)
return None
if kind == "file_name":
if "file_name" in frame.columns:
return frame["file_name"].map(_clean_key)
if "png_path" in frame.columns:
return frame["png_path"].map(
lambda v: (lambda c: None if c is None else os.path.basename(c))(_clean_key(v)))
return None
raise ValueError(f"Unknown join key {kind!r}; expected one of {KEY_PRIORITY}")
def _result_keys(kind: str, results: pd.DataFrame, timelapse: bool) -> Optional[pd.Series]:
"""Build the ``kind`` key for rows of a classifier's results frame."""
if kind == "prcfo":
if "prcfo" in results.columns:
return results["prcfo"].map(_clean_prcfo)
if results.index.name == "prcfo":
return pd.Series(results.index,
index=results.index).map(_clean_prcfo)
name_col = _name_column(results)
if name_col is not None:
return _prcfo_from_names(results[name_col], timelapse)
from .schema import canonicalise_columns
try:
renamed = canonicalise_columns(results.copy())
except Exception: # noqa: BLE001
return None
return _prcfo_from_metadata(renamed)
if kind == "png_path":
for name in ("png_path", "path"):
if name in results.columns:
return results[name].map(_clean_key)
return None
if kind == "file_name":
name_col = _name_column(results)
if name_col is None:
return None
return results[name_col].map(
lambda v: (lambda c: None if c is None else os.path.basename(c))(_clean_key(v)))
raise ValueError(f"Unknown join key {kind!r}; expected one of {KEY_PRIORITY}")
def _choose_key(results: pd.DataFrame, db_frame: pd.DataFrame,
timelapse: bool) -> Tuple[str, pd.Series, pd.Series]:
"""Pick the join key that actually matches the most database rows.
Measured rather than assumed: a key is only better if it lands on more
rows. Ties go to the earliest entry of :data:`KEY_PRIORITY`, which is how
``prcfo`` wins the (normal) case where every candidate matches everything.
:returns: ``(kind, result_keys, db_keys)``.
"""
best = None
for kind in KEY_PRIORITY:
result_keys = _result_keys(kind, results, timelapse)
db_keys = _db_keys(kind, db_frame)
if result_keys is None or db_keys is None:
continue
wanted = set(result_keys.dropna().tolist())
matched = int(db_keys.isin(wanted).sum()) if wanted else 0
if best is None or matched > best[0]:
best = (matched, kind, result_keys, db_keys)
if best is None:
raise ValueError(
"No usable join key: the results frame must carry 'prcfo' (or an "
f"index named 'prcfo'), or one of {_NAME_COLUMNS}, and the target "
"table must carry 'prcfo', 'png_path' or the plate metadata "
"'prcfo' is built from.")
return best[1], best[2], best[3]
@dataclass
[docs]
class MergeReport:
"""What one merge did, in numbers.
:param table: database table into which prediction results were merged.
:param key: join-key strategy selected for the merge.
:param columns: prediction columns requested for insertion or update.
:param db_rows: target-table rows considered by the merge.
:param result_rows: incoming prediction rows considered by the merge.
:param matched_rows: target rows that received at least one prediction.
:param matched_keys: distinct incoming identities found in the target table.
:param unmatched_db_rows: target rows left unchanged because no result
carried their identity.
:param unmatched_result_rows: parseable result rows whose identity was not
present in the target table.
:param unparsed_result_rows: result rows from which no join identity could
be constructed.
:param ambiguous_keys: identities repeated with conflicting prediction
values and therefore deliberately not written.
:param ambiguous_result_rows: incoming rows involved in those conflicts.
:param fanout_rows: additional target rows sharing a matched identity and
receiving the same value, such as alternate crops of one object.
:param repaired: legacy prediction columns repaired before this merge, as
``(table, column, rows_repaired)`` records.
:param added_columns: prediction columns newly created in the target table.
Returned by :func:`merge_prediction_results` and printed by it. Every count
is reported so a partial merge is visible.
"""
table: str
key: str
columns: Tuple[str, ...] = ()
db_rows: int = 0
result_rows: int = 0
#: Database rows that got a value written.
matched_rows: int = 0
#: Distinct keys that matched something.
matched_keys: int = 0
#: Database rows left untouched because no result row had their key.
unmatched_db_rows: int = 0
#: Result rows whose key exists nowhere in the table.
unmatched_result_rows: int = 0
#: Result rows whose name could not be parsed into a key at all.
unparsed_result_rows: int = 0
#: Keys that arrived more than once carrying *different* values. Written
#: nowhere: a wrong score is worse than a missing one.
ambiguous_keys: int = 0
#: Result rows involved in those collisions.
ambiguous_result_rows: int = 0
#: Database rows that shared a key with another row and so were given the
#: same value. Not an error -- one object, several crop modes -- but worth
#: seeing.
fanout_rows: int = 0
#: Legacy columns repaired on the way in, as
#: ``(table, column, rows_repaired)``.
repaired: Tuple[Tuple[str, str, int], ...] = ()
#: Columns created by this merge.
added_columns: Tuple[str, ...] = ()
[docs]
def summary(self) -> str:
"""Return the human-readable multi-line report."""
cols = ", ".join(self.columns)
lines = [
f"Merged {cols} into {self.table} on '{self.key}': "
f"{self.matched_rows}/{self.db_rows} rows matched"
]
if self.db_rows and not self.matched_rows:
lines.append(
f" !! NOTHING MATCHED. {self.result_rows} result row(s) and "
f"{self.db_rows} {self.table} row(s) share no '{self.key}' "
f"value, so no score was written. The results probably come "
f"from a different experiment than {self.table}.")
for label, value in (
("result row(s) had no matching row in this database",
self.unmatched_result_rows),
("result row(s) had a name no key could be parsed from",
self.unparsed_result_rows),
(f"{self.table} row(s) were left unscored", self.unmatched_db_rows),
(f"{self.table} row(s) shared a key with another row and got the "
f"same value", self.fanout_rows),
):
if value:
lines.append(f" {value} {label}")
if self.ambiguous_keys:
lines.append(
f" !! {self.ambiguous_keys} key(s) arrived from "
f"{self.ambiguous_result_rows} result rows with conflicting "
f"values and were NOT written. Two crops sharing a '{self.key}' "
f"cannot be told apart -- this happens when two source folders "
f"give their plates the same name.")
for _table, column, count in self.repaired:
lines.append(
f" repaired {count} legacy row(s) of '{column}' (the Annotate "
f"app's 1/2 class encoding back to the model's 0/1)")
return "\n".join(lines)
[docs]
def __str__(self) -> str:
"""Return the same human-readable report as :meth:`summary`.
:returns: Multi-line prediction-merge summary.
"""
return self.summary()
def _execute(cursor, sql: str) -> None:
"""Run one write statement.
A named seam, like :func:`_execute_updates`, so a test can make the write
fail after it has taken effect and prove the transaction rolls it back.
"""
cursor.execute(sql)
[docs]
def migrate_prediction_columns(db_path, table: str = PNG_TABLE,
verbose: bool = True) -> List[Tuple[str, str, int]]:
"""Put a legacy prediction column back into the encoding it claims to be in.
:func:`spacr.utils.add_column_to_database` --- the ML stage's old write
path --- replaced every ``0`` with a ``2`` on the way into the database,
because the Annotate app labels classes 1 and 2. Nothing else did that, so
``png_list.predictions`` said ``2`` where ``results.csv`` from the very same
run said ``0``: one number, two meanings, depending on which file you
opened. This repairs it, so a database written before the change reads
correctly with no manual action --- the same repair-on-read contract
:func:`spacr.utils.rename_columns_in_db` has, and the same properties:
* **Idempotent.** After the repair the column holds 0s and 1s, which is not
the encoding this looks for, so a second pass does nothing.
* **Never destructive.** The substitution is only reversed when the column
holds nothing but 1s and 2s. A three-class model's genuine class 2 is
indistinguishable from a mangled 0, and guessing would be worse than
leaving it exactly as it is.
* **All or nothing.** SQLite, DuckDB and PostgreSQL repair in a
transaction; Parquet publishes a complete replacement part snapshot.
A half-repaired column would be a column in two encodings at once.
:param db_path: SQLite or DuckDB file, Parquet store directory, or
PostgreSQL connection string. A missing local store is a no-op.
:param table: table to migrate. Default ``'png_list'``.
:param verbose: print each repair.
:returns: list of ``(table, column, rows_repaired)``.
"""
from .tabular import _backend_of, resolve_path
backend = _backend_of(db_path)
if backend in ('duckdb', 'parquet'):
db_path = resolve_path(db_path)
if backend != 'postgres' and not os.path.exists(str(db_path)):
return []
if backend in ('duckdb', 'postgres'):
repaired = _migrate_sql_store(str(db_path), table, backend)
if verbose:
for _table, column, count in repaired:
print(f"Repaired {count} row(s) of `{_table}`.`{column}`: the "
f"Annotate app's 1/2 class encoding back to the model's 0/1")
return repaired
if backend == 'parquet':
repaired = _merge_parquet_store(
None, str(db_path), {}, table, 'auto', None,
migration_only=True)
if verbose:
for _table, column, count in repaired:
print(f"Repaired {count} row(s) of `{_table}`.`{column}`: the "
f"Annotate app's 1/2 class encoding back to the model's 0/1")
return repaired
repaired: List[Tuple[str, str, int]] = []
con = sqlite3.connect(str(db_path), timeout=30)
con.isolation_level = None
try:
cur = con.cursor()
tables = {row[0] for row in
cur.execute("SELECT name FROM sqlite_master WHERE type='table'")}
if table not in tables:
return []
cols = [row[1] for row in cur.execute(f"PRAGMA table_info({_quote(table)})")]
pending = [c for c in ANNOTATE_ENCODED_COLUMNS if c in cols]
if not pending:
return []
cur.execute("BEGIN")
try:
for column in pending:
quoted = _quote(column)
distinct = {row[0] for row in cur.execute(
f"SELECT DISTINCT {quoted} FROM {_quote(table)} "
f"WHERE {quoted} IS NOT NULL")}
if not distinct or not distinct.issubset({1, 2}) or 2 not in distinct:
continue
_execute(cur, f"UPDATE {_quote(table)} SET {quoted} = 0 "
f"WHERE {quoted} = 2")
repaired.append((table, column, cur.rowcount))
cur.execute("COMMIT")
except BaseException:
cur.execute("ROLLBACK")
raise
finally:
con.close()
if verbose:
for _table, column, count in repaired:
print(f"Repaired {count} row(s) of `{_table}`.`{column}`: the "
f"Annotate app's 1/2 class encoding back to the model's 0/1")
return repaired
def _execute_updates(cursor, sql: str, updates: Sequence[Tuple]) -> None:
"""Apply the prepared UPDATE statements.
A named seam rather than an inline ``executemany`` so a test can make the
write fail halfway and prove the transaction rolls the whole merge back.
"""
cursor.executemany(sql, updates)
[docs]
def merge_prediction_results(results, db_path, columns, table: str = PNG_TABLE,
key: str = "auto", timelapse: Optional[bool] = None,
verbose: bool = True) -> Optional[MergeReport]:
"""Write a classifier's per-object results onto the rows of ``table``.
Shared by Classify (CV) and Classify (ML) --- see the module docstring for
why the key is ``prcfo``, why an ambiguous key is refused rather than
guessed at, and why the whole thing is one transaction.
:param results: DataFrame of per-object results. Must carry the source
columns named in ``columns``, plus something to key on: a ``prcfo``
column, an index named ``prcfo``, or a ``path``/``png_path``/
``file_name`` column holding spaCR crop names.
:param db_path: SQLite or DuckDB file, Parquet store directory, or
PostgreSQL connection string. A missing local store is reported and
skipped.
:param columns: mapping of *database* column name to
``(results column, 'REAL' | 'INTEGER')``.
:param table: target table. Default ``'png_list'``.
:param key: ``'auto'`` (measure every candidate and use the best) or one of
:data:`KEY_PRIORITY` to force it.
:param timelapse: whether crop names carry a timepoint. ``None`` detects it
from the presence of a time column on the table.
:param verbose: print the report.
:returns: a :class:`MergeReport`, or ``None`` if a local store is missing.
:raises KeyError: if ``results`` lacks one of the source columns.
:raises ValueError: if a Parquet table is missing or prediction output
would replace a join-key column in an alternate store.
"""
from .tabular import _backend_of, resolve_path
backend = _backend_of(db_path)
db_path = (resolve_path(db_path) if backend in ('duckdb', 'parquet')
else str(db_path))
if backend != 'postgres' and not os.path.exists(db_path):
print(f"Database not found at {db_path}; skipping merge.")
return None
if not isinstance(results, pd.DataFrame):
results = pd.DataFrame(results)
spec = {db_col: (src if isinstance(src, (tuple, list)) else (src, "REAL"))
for db_col, src in columns.items()}
missing = [src for src, _type in spec.values() if src not in results.columns]
if missing:
raise KeyError(
f"merge_prediction_results: results frame has no column(s) "
f"{missing}; it carries {list(results.columns)[:12]}"
+ (" ..." if len(results.columns) > 12 else ""))
if backend in ('duckdb', 'postgres'):
report = _merge_sql_store(results, db_path, spec, table, key,
timelapse, backend)
if verbose:
print(report.summary())
return report
if backend == 'parquet':
report = _merge_parquet_store(results, db_path, spec, table, key,
timelapse)
if verbose:
print(report.summary())
return report
repaired = migrate_prediction_columns(db_path, table=table, verbose=False)
con = sqlite3.connect(db_path, timeout=30)
con.isolation_level = None
try:
cur = con.cursor()
cur.execute("BEGIN")
try:
report = _merge_locked(cur, results, spec, table, key, timelapse,
repaired)
cur.execute("COMMIT")
except BaseException:
cur.execute("ROLLBACK")
raise
finally:
con.close()
if verbose:
print(report.summary())
return report
def _repair_sql_columns(conn, table: str, table_columns: Sequence[str]):
"""Repair the legacy 1/2 class encoding inside the caller's transaction."""
quoted_table = _quote(table)
repaired = []
for column in ANNOTATE_ENCODED_COLUMNS:
if column not in table_columns:
continue
quoted = _quote(column)
values = {row[0] for row in conn.execute(
f"SELECT DISTINCT {quoted} FROM {quoted_table} "
f"WHERE {quoted} IS NOT NULL").fetchall()}
if not values or not values.issubset({1, 2}) or 2 not in values:
continue
count = conn.execute(
f"SELECT COUNT(*) FROM {quoted_table} WHERE {quoted} = 2"
).fetchone()[0]
conn.execute(f"UPDATE {quoted_table} SET {quoted} = 0 WHERE {quoted} = 2")
repaired.append((table, column, int(count)))
return repaired
def _migrate_sql_store(db_path, table, backend):
"""Repair a native SQL store in one transaction, if its table exists."""
from .tabular import _duckdb_connect, _postgres_connect
connect = _duckdb_connect if backend == 'duckdb' else _postgres_connect
conn = connect(db_path)
try:
if backend == 'duckdb':
conn.execute('BEGIN TRANSACTION')
try:
present = {row[0] for row in conn.execute(
'SELECT table_name FROM information_schema.tables '
'WHERE table_schema = current_schema() '
"AND table_type = 'BASE TABLE'").fetchall()}
if table in present:
quoted = _quote(table)
if backend == 'postgres':
conn.execute(f'LOCK TABLE {quoted} IN EXCLUSIVE MODE')
columns = [getattr(entry, 'name', None) or entry[0]
for entry in conn.execute(
f'SELECT * FROM {quoted} LIMIT 0').description]
repaired = _repair_sql_columns(conn, table, columns)
else:
repaired = []
if backend == 'duckdb':
conn.execute('COMMIT')
else:
conn.commit()
except BaseException:
if backend == 'duckdb':
conn.execute('ROLLBACK')
else:
conn.rollback()
raise
finally:
conn.close()
return repaired
def _execute_native_update(conn, sql, params):
"""Apply one native-store row update inside the caller's transaction."""
conn.execute(sql, params)
def _merge_sql_store(results, db_path, spec, table, key, timelapse, backend):
"""Update DuckDB or PostgreSQL rows while holding one native transaction."""
from .tabular import _duckdb_connect, _postgres_connect
connect = _duckdb_connect if backend == 'duckdb' else _postgres_connect
conn = connect(db_path)
quoted_table = _quote(table)
placeholder = '?' if backend == 'duckdb' else '%s'
try:
if backend == 'duckdb':
conn.execute('BEGIN TRANSACTION')
else:
conn.execute(f'LOCK TABLE {quoted_table} IN EXCLUSIVE MODE')
try:
table_columns = [getattr(entry, 'name', None) or entry[0]
for entry in conn.execute(
f'SELECT * FROM {quoted_table} LIMIT 0').description]
key_columns = [column for column in table_columns
if column in ('prcfo', 'png_path', 'file_name')
or column in _PRCFO_METADATA
or column in _OBJECT_ID_COLUMNS
or column in ('timeID', 'time_id')]
if set(spec) & set(key_columns):
raise ValueError('Prediction output cannot replace a join-key column.')
query = ', '.join(_quote(column) for column in key_columns)
rows = conn.execute(
f'SELECT {query} FROM {quoted_table}').fetchall() if query else []
db_frame = pd.DataFrame(rows, columns=key_columns)
repaired = _repair_sql_columns(conn, table, table_columns)
updates, report, order, types = _plan_merge(
results, spec, table, key, timelapse, table_columns,
db_frame, range(len(db_frame)), repaired)
native_types = {'REAL': 'DOUBLE PRECISION',
'INTEGER': 'BIGINT', 'TEXT': 'TEXT'}
for column in report.added_columns:
sql_type = types[column].upper()
if sql_type not in native_types:
raise ValueError(f'Unsupported prediction SQL type: {sql_type!r}')
conn.execute(f'ALTER TABLE {quoted_table} ADD COLUMN '
f'{_quote(column)} {native_types[sql_type]}')
if updates:
assignment = ', '.join(f'{_quote(column)} = {placeholder}'
for column in order)
selector = ' AND '.join(
f'{_quote(column)} IS NOT DISTINCT FROM {placeholder}'
for column in key_columns)
sql = f'UPDATE {quoted_table} SET {assignment} WHERE {selector}'
for update in updates:
raw = db_frame.iloc[update[-1]]
params = tuple(update[:-1]) + tuple(
_sql_value(raw[column], 'TEXT') for column in key_columns)
_execute_native_update(conn, sql, params)
if backend == 'duckdb':
conn.execute('COMMIT')
else:
conn.commit()
except BaseException:
if backend == 'duckdb':
conn.execute('ROLLBACK')
else:
conn.rollback()
raise
finally:
conn.close()
return report
def _merge_parquet_store(results, db_path, spec, table, key, timelapse,
migration_only=False):
"""Publish scored Parquet parts through one active-snapshot replacement."""
from bisect import bisect_right
import uuid
from .tabular import (
_parquet_manifest, _parquet_part_record, _parquet_parts,
_parquet_store_lock, _parquet_table_folder, _publish,
_publish_parquet_manifest, _require_optional, _PYARROW_MISSING_MESSAGE,
)
arrow = _require_optional('pyarrow', _PYARROW_MISSING_MESSAGE)
parquet = _require_optional('pyarrow.parquet', _PYARROW_MISSING_MESSAGE)
if table in ('.', '..') or '/' in table or '\\' in table:
raise ValueError(f'Invalid Parquet table name: {table!r}')
folder = _parquet_table_folder(db_path, table)
with _parquet_store_lock(db_path):
parts = _parquet_parts(db_path, table)
if not parts:
if migration_only:
return []
raise ValueError(f'Table not found in database: {table}')
manifest = _parquet_manifest(folder)
records = (manifest['active'] if manifest is not None else
[_parquet_part_record(part) for part in parts])
retired = manifest['retired'] if manifest is not None else []
schemas = [parquet.read_schema(part) for part in parts]
table_columns = list(dict.fromkeys(
name for schema in schemas for name in schema.names))
key_columns = [column for column in table_columns
if column in ('prcfo', 'png_path', 'file_name')
or column in _PRCFO_METADATA
or column in _OBJECT_ID_COLUMNS
or column in ('timeID', 'time_id')]
if set(spec) & set(key_columns):
raise ValueError('Prediction output cannot replace a join-key column.')
key_frames = []
offsets = [0]
repair_values = set()
repair_count = 0
for part, part_schema in zip(parts, schemas):
present = [column for column in key_columns
if column in part_schema.names]
selected = parquet.read_table(part, columns=present).to_pandas()
key_frames.append(selected.reindex(columns=key_columns))
offsets.append(offsets[-1] + len(selected))
if ML_CLASS_COLUMN in part_schema.names:
values = parquet.read_table(
part, columns=[ML_CLASS_COLUMN]).column(0).to_pylist()
repair_values.update(value for value in values if value is not None)
repair_count += sum(value == 2 for value in values)
should_repair = (bool(repair_values) and
repair_values.issubset({1, 2}) and 2 in repair_values)
repaired = ([(table, ML_CLASS_COLUMN, repair_count)]
if should_repair else [])
if migration_only:
updates, report, order, types = [], None, [], {}
else:
db_frame = pd.concat(key_frames, ignore_index=True)
updates, report, order, types = _plan_merge(
results, spec, table, key, timelapse, table_columns,
db_frame, range(len(db_frame)), repaired)
if not updates and not should_repair and (
migration_only or not report.added_columns):
return repaired if migration_only else report
sql_to_arrow = {'REAL': arrow.float64(), 'INTEGER': arrow.int64(),
'TEXT': arrow.string()}
if not migration_only:
for column in report.added_columns:
if types[column].upper() not in sql_to_arrow:
raise ValueError(
f'Unsupported prediction SQL type: {types[column]!r}')
grouped = [[] for _ in parts]
for update in updates:
part_index = bisect_right(offsets, update[-1]) - 1
grouped[part_index].append((update[-1] - offsets[part_index],
update[:-1]))
staged = []
published = False
try:
for part, part_updates in zip(parts, grouped):
source = parquet.read_table(part)
changed = source
for position, column in enumerate(order):
if column in changed.schema.names:
field = changed.schema.field(column)
values = changed.column(column).to_pylist()
else:
field = arrow.field(column, sql_to_arrow[types[column].upper()])
values = [None] * changed.num_rows
if should_repair and column == ML_CLASS_COLUMN:
values = [0 if value == 2 else value for value in values]
for row, update_values in part_updates:
values[row] = update_values[position]
array = arrow.array(values, type=field.type)
if column in changed.schema.names:
changed = changed.set_column(
changed.schema.get_field_index(column), field, array)
else:
changed = changed.append_column(field, array)
if should_repair and ML_CLASS_COLUMN not in order \
and ML_CLASS_COLUMN in changed.schema.names:
field = changed.schema.field(ML_CLASS_COLUMN)
values = changed.column(ML_CLASS_COLUMN).to_pylist()
values = [0 if value == 2 else value for value in values]
changed = changed.set_column(
changed.schema.get_field_index(ML_CLASS_COLUMN),
field, arrow.array(values, type=field.type))
staged_part = os.path.join(
folder, f'.spacr-part-{uuid.uuid4().hex}.parquet')
_publish(staged_part, lambda pending, value=changed:
parquet.write_table(value, pending))
staged.append(staged_part)
if (_parquet_manifest(folder) != manifest or
_parquet_parts(db_path, table) != parts or
[_parquet_part_record(part) for part in parts] != records):
raise RuntimeError('Parquet target changed while scoring; retry.')
next_active = [_parquet_part_record(part) for part in staged]
_publish_parquet_manifest(folder, next_active, retired + records)
published = True
finally:
if not published:
for part in staged:
if os.path.exists(part):
os.remove(part)
return repaired if migration_only else report
def _merge_locked(cur, results: pd.DataFrame, spec: Mapping[str, Tuple[str, str]],
table: str, key: str, timelapse: Optional[bool],
repaired: Sequence[Tuple[str, str, int]]) -> MergeReport:
"""Do the merge inside an already-open transaction."""
quoted_table = _quote(table)
cur.execute(f"SELECT * FROM {quoted_table} LIMIT 0")
table_columns = [d[0] for d in cur.description]
key_columns = [c for c in table_columns
if c in ("prcfo", "png_path", "file_name")
or c in _PRCFO_METADATA or c in _OBJECT_ID_COLUMNS
or c in ("timeID", "time_id")]
rowid = _rowid_alias(table_columns)
select = ", ".join([rowid] + [_quote(c) for c in key_columns])
rows = cur.execute(f"SELECT {select} FROM {quoted_table}").fetchall()
rowids = [r[0] for r in rows]
db_frame = pd.DataFrame([r[1:] for r in rows], columns=key_columns)
updates, report, order, types = _plan_merge(
results, spec, table, key, timelapse, table_columns,
db_frame, rowids, repaired)
for db_col in report.added_columns:
cur.execute(f"ALTER TABLE {quoted_table} ADD COLUMN "
f"{_quote(db_col)} {types[db_col]}")
if updates:
assignments = ", ".join(f"{_quote(c)} = ?" for c in order)
_execute_updates(
cur, f"UPDATE {quoted_table} SET {assignments} WHERE {rowid} = ?",
updates)
return report
def _plan_merge(results: pd.DataFrame, spec: Mapping[str, Tuple[str, str]],
table: str, key: str, timelapse: Optional[bool],
table_columns: Sequence[str], db_frame: pd.DataFrame,
rowids: Sequence, repaired: Sequence[Tuple[str, str, int]]):
"""Plan row updates and counts without depending on a store's SQL dialect."""
from .utils import _time_column
if timelapse is None:
timelapse = _time_column(table_columns) is not None
if key == "auto":
kind, result_keys, db_keys = _choose_key(results, db_frame, timelapse)
else:
kind = key
result_keys = _result_keys(kind, results, timelapse)
db_keys = _db_keys(kind, db_frame)
if result_keys is None or db_keys is None:
raise ValueError(
f"Join key {kind!r} cannot be built: the results frame or "
f"{table} does not carry what it is made of.")
value_frames = {db_col: results[src] for db_col, (src, _t) in spec.items()}
types = {db_col: sql_type for db_col, (_src, sql_type) in spec.items()}
order = list(spec)
lookup: Dict[str, Tuple] = {}
conflicting: Dict[str, int] = {}
unparsed = 0
key_list = [_clean_key(value) for value in result_keys]
db_key_list = [_clean_key(value) for value in db_keys]
columns_by_row = [list(value_frames[db_col]) for db_col in order]
for idx, row_key in enumerate(key_list):
if row_key is None:
unparsed += 1
continue
values = tuple(_sql_value(columns_by_row[c][idx], types[order[c]])
for c in range(len(order)))
previous = lookup.get(row_key, _MISSING)
if previous is _MISSING:
lookup[row_key] = values
elif not _values_equal(previous, values):
conflicting[row_key] = conflicting.get(row_key, 1) + 1
for row_key in conflicting:
lookup.pop(row_key, None)
added = [db_col for db_col in order if db_col not in table_columns]
updates = []
matched_keys = set()
for position, row_key in enumerate(db_key_list):
values = lookup.get(row_key) if row_key is not None else None
if values is None:
continue
updates.append(tuple(values) + (rowids[position],))
matched_keys.add(row_key)
db_key_set = {k for k in db_key_list if k is not None}
unmatched_results = sum(
1 for k in key_list
if k is not None and k not in conflicting and k not in db_key_set)
report = MergeReport(
table=table,
key=kind,
columns=tuple(order),
db_rows=len(rowids),
result_rows=len(key_list),
matched_rows=len(updates),
matched_keys=len(matched_keys),
unmatched_db_rows=len(rowids) - len(updates),
unmatched_result_rows=unmatched_results,
unparsed_result_rows=unparsed,
ambiguous_keys=len(conflicting),
ambiguous_result_rows=sum(conflicting.values()),
fanout_rows=len(updates) - len(matched_keys),
repaired=tuple(repaired),
added_columns=tuple(added),
)
return updates, report, order, types
[docs]
def attach_predictions(objects, results, *,
score_source: str = "pred",
class_source: str = "cv_predictions",
score_col: str = CV_SCORE_COLUMN,
class_col: str = CV_CLASS_COLUMN,
timelapse: bool = False):
"""Join prediction columns onto an object frame in memory.
The same join as :func:`merge_prediction_results`, without writing
anything. `_choose_key` picks the key that matches the most rows, so a
montage reading scores from a score CSV and a database with the same CSV
merged into it assign each object the same score.
This supports projects whose ``png_list`` table has no prediction column
but whose regression inputs already contain one score row per cell. The
join uses those scores without changing the database.
:param objects: the per-object frame, e.g. `png_list` read back.
:param results: the score table -- `path`, `pred`, `cv_predictions`, as
`process_vision_results` and the regression module's score CSVs carry.
:returns: ``(frame, matched)`` -- a copy of ``objects`` with the score and
class columns added where they joined, and how many rows matched.
``matched`` is 0 when nothing lined up, and the frame comes back
without the columns, so a caller can report the count.
"""
if objects is None or results is None or not len(objects) or not len(results):
return objects, 0
try:
kind, result_keys, db_keys = _choose_key(results, objects, timelapse)
except ValueError:
return objects, 0
out = objects.copy()
score_name = (str(score_source)
if str(score_source) in getattr(results, "columns", ())
else first_present(results, SCORE_SOURCE_COLUMNS))
class_name = (str(class_source)
if str(class_source) in getattr(results, "columns", ())
else first_present(results, CLASS_SOURCE_COLUMNS))
wanted = {score_col: score_name, class_col: class_name}
matched = 0
for target, source in wanted.items():
if source is None or source not in getattr(results, "columns", ()):
continue
lookup = dict(zip(result_keys, results[source]))
joined = db_keys.map(lambda key: lookup.get(key))
found = int(joined.notna().sum())
if not found:
continue
out[target] = joined
matched = max(matched, found)
return (out, matched) if matched else (objects, 0)
[docs]
def merge_cv_predictions(df, db_path, table: str = PNG_TABLE,
score_col: str = CV_SCORE_COLUMN,
class_col: str = CV_CLASS_COLUMN,
score_source: str = "pred",
class_source: str = "cv_predictions",
verbose: bool = True) -> Optional[MergeReport]:
"""Merge :func:`spacr.deep_spacr.apply_model_to_tar` results into ``table``.
:param df: frame from ``apply_model_to_tar`` -> ``process_vision_results``
(``path``, ``pred``, ``cv_predictions``).
:param db_path: SQLite database to write into.
:param table: target table. Default ``'png_list'``.
:param score_col: database column for the probability.
:param class_col: database column for the thresholded class.
:param score_source: column of ``df`` the probability comes from.
:param class_source: column of ``df`` the class comes from.
:param verbose: print the report.
:returns: a :class:`MergeReport`, or ``None`` if the database is missing.
"""
return merge_prediction_results(
df, db_path,
{score_col: (score_source, "REAL"), class_col: (class_source, "INTEGER")},
table=table, verbose=verbose)
[docs]
def merge_ml_predictions(df, db_path, table: str = PNG_TABLE,
score_col: str = ML_SCORE_COLUMN,
class_col: str = ML_CLASS_COLUMN,
verbose: bool = True) -> Optional[MergeReport]:
"""Merge :func:`spacr.ml.ml_analysis` results into ``table``.
``ml_analysis`` returns the predicted class in ``predictions`` and the
per-class probabilities in ``prediction_probability_class_<i>``; the
positive-class probability is taken from class 1 when the model produced
one, and skipped when it did not (a one-class fit) rather than inventing a
column.
:param df: the scored frame -- ``ml_analysis`` output ``[0]``.
:param db_path: SQLite database to write into.
:param table: target table. Default ``'png_list'``.
:param score_col: database column for the probability.
:param class_col: database column for the predicted class.
:param verbose: print the report.
:returns: a :class:`MergeReport`, or ``None`` if the database is missing,
or if the frame carries no prediction column at all.
"""
columns: Dict[str, Tuple[str, str]] = {}
if "predictions" in df.columns:
columns[class_col] = ("predictions", "INTEGER")
if "prediction_probability_class_1" in df.columns:
columns[score_col] = ("prediction_probability_class_1", "REAL")
if not columns:
print("No prediction columns on the ML results frame; skipping merge "
f"into {db_path}.")
return None
return merge_prediction_results(df, db_path, columns, table=table,
verbose=verbose)