"""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, and both used to leave those rows in a CSV next
to the model. A CSV is not where the rest of spaCR looks: 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 have
to land there, on the row they belong to.
Why this is its own module
--------------------------
The merge used to live in :mod:`spacr.deep_spacr`, which is an odd home for
something :mod:`spacr.ml` needs just as much --- importing ``deep_spacr`` pulls
in torch and torchvision, which the classical-ML path has no use for. The
obvious alternative, :mod:`spacr.utils`, already owns the database *write*
helpers (``_append_to_measurements_db``, ``rename_columns_in_db``), but it is
also 8000 lines and imports most of the package; adding a third database
concern there would make it harder, not easier, to see that CV and ML share one
code path. 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 land on *the same row*, which is the whole point of letting
them coexist.
The previous implementation matched on ``os.path.basename(png_path)``. A
basename 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
* the old lookup was a plain ``dict`` assignment, so the second of those two
crops silently overwrote the first and one of the two plates was scored with
the other plate's predictions.
The second half of that is the real defect, and it is not fixed by changing the
key: two crops that share a basename share a ``prcfo`` too. So the fix is to
*detect* the collision. 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
actually matches the most rows wins, ties going to the earliest in
:data:`KEY_PRIORITY`. The chosen key and every count are printed, because a
merge that matched 3 of 40000 rows used to look exactly like 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 here because a merge that matched three rows of forty thousand
used to be indistinguishable from one that matched all of them.
"""
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, _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 = os.path.join(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`, AND NOTHING WRITTEN.
`_choose_key` picks the key by MEASURING which one lands on the most rows,
so a montage reading scores out of a score CSV and a database that had the
same CSV merged into it cannot disagree about which object got which
number -- which two separate join implementations eventually would.
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 refuse with a real number.
"""
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)