"""Build an annotation set by streaming crops, and register it for annotating.
Annotation works on single-object crops. Producing them with Measure means
running it over a whole plate -- about twenty minutes for 52 fields on a
30-core machine -- and running it again to cut the crops differently.
Every field's objects are already described twice over once Measure has run:
by the label masks inside the merged arrays, and by the coordinate columns in
``measurements.db``. Either is enough to cut crops from, so a second set can be
built in seconds without measuring anything again.
The two routes produce different crops:
* ``array`` reads the object masks out of the merged stacks, so it can cut to
the object itself -- masked, or to its bounding box;
* ``database`` reads the coordinate columns, which are all the database
stores, so it can only produce a bounding box.
Comparing the two is therefore only meaningful with ``bounding_box=True``.
:func:`spacr.annotation_dataset` says so in the settings it accepts, and the
GUI says it beside the picker.
"""
from __future__ import annotations
import logging
import os
import re
import sqlite3
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
import pandas as pd
from .database_concurrency import connect as connect_database
from .tabular import read_database
LOG = logging.getLogger("spacr.annotation_dataset")
__all__ = [
"PNG_TABLE_BASE",
"STREAM_SOURCES",
"next_png_table",
"filter_selection",
"png_list_frame",
"write_png_list",
"generate_annotation_dataset",
]
#: The table the annotation viewer reads. See `spacr.agreement.PNG_TABLE`.
PNG_TABLE_BASE = "png_list"
#: Where the objects can be read from, as ``(value, label)``.
#:
#: The names match `stream_dataset.STREAM_METHODS` in meaning but not in
#: spelling -- that module says "column" for what a user calls the database.
STREAM_SOURCES: Tuple[Tuple[str, str], ...] = (
("array", "the object masks in the merged arrays"),
("database", "the coordinate columns in measurements.db"),
)
#: ``png_list`` as `measure_crop` writes it. Matching it exactly is what lets
#: the annotation viewer open a streamed set without knowing how it was made.
PNG_LIST_COLUMNS: Tuple[str, ...] = (
"png_path", "file_name", "plateID", "rowID", "columnID", "fieldID",
"prcfo", "cell_id", "annotate",
)
_TABLE_SUFFIX = re.compile(r"^png_list(?:_(\d+))?$")
[docs]
def next_png_table(connection: sqlite3.Connection) -> str:
"""Return the table name a new annotation set should be written under.
``png_list`` when it is free, then ``png_list_2``, ``png_list_3`` and so
on. An existing set is never overwritten: it may already carry
annotations, and those are hand-made and unrecoverable.
The caller must hold a write transaction. Choosing a name and creating the
table are two steps, and two runs started together would otherwise choose
the same one. :func:`write_png_list` does this correctly; call it rather
than this.
:param connection: an open connection to the measurements database.
:returns: the free table name.
"""
taken = set()
for (name,) in connection.execute(
"select name from sqlite_master where type in ('table', 'view')"):
match = _TABLE_SUFFIX.match(str(name))
if match:
taken.add(int(match.group(1) or 1))
number = 1
while number in taken:
number += 1
return PNG_TABLE_BASE if number == 1 else f"{PNG_TABLE_BASE}_{number}"
[docs]
def filter_selection(selection: pd.DataFrame,
settings: Mapping[str, Any]) -> pd.DataFrame:
"""Drop objects a run would not have cropped.
The predicates are the ones `measure_crop` applies, so a streamed set and
a measured one describe the same population. They are expressed against the
selection table's columns rather than against a measurement frame, because
the array route has no measurements -- only labels and geometry.
Recognised settings, each optional and each skipped when absent:
``{object}_min_size`` / ``{object}_max_size``
Object area in pixels. ``0`` and ``None`` both mean "no bound", which
is what the Measure panel writes for an unset field -- treating 0 as a
real minimum would drop nothing and look like it had worked.
``wells`` / ``exclude_wells``
Well ids to keep or drop, as ``rowID+columnID`` pairs or as plain
well names.
``max_objects``
A cap, applied LAST and deterministically (by the sort the selection
already has), so a capped set is reproducible.
:param selection: a selection table from :mod:`spacr.stream_dataset`.
:param settings: the run settings.
:returns: a new frame; the input is not modified.
"""
if selection is None or not len(selection):
return selection
frame = selection.copy()
object_type = str(settings.get("object_array") or "cell")
area = None
for candidate in ("area", f"{object_type}_area", "object_area"):
if candidate in frame.columns:
area = candidate
break
if area is not None:
minimum = settings.get(f"{object_type}_min_size")
maximum = settings.get(f"{object_type}_max_size")
if minimum:
frame = frame[frame[area] >= float(minimum)]
if maximum:
frame = frame[frame[area] <= float(maximum)]
keep = settings.get("wells")
drop = settings.get("exclude_wells")
if keep or drop:
names = _well_names(frame)
if keep:
frame = frame[names.isin({str(w) for w in keep})]
names = _well_names(frame)
if drop:
frame = frame[~names.isin({str(w) for w in drop})]
cap = settings.get("max_objects")
if cap:
frame = frame.head(int(cap))
return frame.reset_index(drop=True)
def _well_names(frame: pd.DataFrame) -> pd.Series:
"""Well ids as one comparable string per row."""
if "well" in frame.columns:
return frame["well"].astype(str)
row = frame["rowID"].astype(str) if "rowID" in frame.columns else ""
column = frame["columnID"].astype(str) if "columnID" in frame.columns else ""
return (row + column) if len(frame) else pd.Series([], dtype=str)
[docs]
def png_list_frame(selection: pd.DataFrame, paths: Sequence[str]) -> pd.DataFrame:
"""The rows the annotation viewer reads, in `measure_crop`'s own schema.
:param selection: the filtered selection table.
:param paths: the written crop path for each of its rows, in order.
:returns: a frame with exactly :data:`PNG_LIST_COLUMNS`.
"""
if selection is None or not len(selection):
return pd.DataFrame(columns=list(PNG_LIST_COLUMNS))
frame = selection.reset_index(drop=True).copy()
if len(paths) != len(frame):
raise ValueError(
f"{len(paths)} crop paths for {len(frame)} selected objects; the "
"two must correspond row for row or the table would name the "
"wrong picture for an object")
out = pd.DataFrame({
"png_path": [str(p) for p in paths],
"file_name": [os.path.basename(str(p)) for p in paths],
"plateID": frame.get("plateID", ""),
"rowID": frame.get("rowID", ""),
"columnID": frame.get("columnID", ""),
"fieldID": frame.get("fieldID", ""),
"cell_id": ["o" + str(v) for v in frame.get("objectID", [])],
})
out["fieldID"] = out["fieldID"].astype(str)
out["prcfo"] = (out["plateID"].astype(str) + "_"
+ out["rowID"].astype(str) + "_"
+ out["columnID"].astype(str) + "_f"
+ out["fieldID"].str.lstrip("f") + "_"
+ out["cell_id"].astype(str))
out["annotate"] = None
return out[list(PNG_LIST_COLUMNS)]
[docs]
def reserve_png_table(db_path: str) -> str:
"""Claim the next free table name by creating it, empty.
The name is reserved before the crops are cut, so the folder they are
written into can be named to match: `png_list_2` gets `data_2`. Deriving
the folder from the table is what makes a set on disk traceable to the set
in the database -- with two independent counters they drift the first time
either is deleted, and then nothing says which folder a table describes.
Creating the table is what reserves it: two runs started together would
otherwise choose the same name, and the second would fail on insert after
it had already written a folder full of crops.
:param db_path: the measurements database.
:returns: the reserved table name.
"""
connection = connect_database(str(db_path))
try:
connection.execute("BEGIN IMMEDIATE")
name = next_png_table(connection)
columns = ", ".join(f'"{c}"' for c in PNG_LIST_COLUMNS)
connection.execute(f'CREATE TABLE "{name}" ({columns})')
connection.execute("COMMIT")
except Exception:
try:
connection.execute("ROLLBACK")
except sqlite3.Error:
pass
raise
finally:
connection.close()
return name
[docs]
def crops_folder_for(table: str) -> str:
"""The crop folder that belongs to ``table``.
``png_list`` -> ``data``; ``png_list_2`` -> ``data_2``. The suffix is
carried across rather than counted again, so the pair cannot drift.
"""
suffix = str(table)[len(PNG_TABLE_BASE):]
return "data" + suffix
[docs]
def write_png_list(db_path: str, frame: pd.DataFrame, *,
table: Optional[str] = None) -> str:
"""Write an annotation set into the measurements database.
The name is chosen and the table created inside one transaction, so two
runs started together cannot pick the same one. A caller that already
reserved a name with :func:`reserve_png_table` passes it as ``table``.
:param db_path: the measurements database.
:param frame: rows as :func:`png_list_frame` builds them.
:param table: a name already reserved.
:returns: the table actually written.
"""
connection = connect_database(str(db_path))
try:
connection.execute("BEGIN IMMEDIATE")
name = table or next_png_table(connection)
columns = ", ".join(f'"{c}"' for c in PNG_LIST_COLUMNS)
if table is None:
connection.execute(f'CREATE TABLE "{name}" ({columns})')
placeholders = ", ".join("?" for _ in PNG_LIST_COLUMNS)
connection.executemany(
f'INSERT INTO "{name}" ({columns}) VALUES ({placeholders})',
[tuple(None if pd.isna(v) else v for v in row)
for row in frame[list(PNG_LIST_COLUMNS)].itertuples(index=False)])
connection.execute("COMMIT")
except Exception:
try:
connection.execute("ROLLBACK")
except sqlite3.Error:
pass
raise
finally:
connection.close()
LOG.info("wrote %d rows to %s in %s", len(frame), name, db_path)
return name
[docs]
def generate_annotation_dataset(settings: Mapping[str, Any]) -> Dict[str, Any]:
"""Stream a set of crops and register it for annotation.
:param settings: needs ``src`` (the plate folder) and accepts
``stream_source`` (:data:`STREAM_SOURCES`), ``object_array``,
``channel_arrays``, ``bounding_box``, the filtration keys
:func:`filter_selection` reads, and ``dst`` for where the crops go.
:returns: the streaming report, plus ``table`` naming what was written.
"""
from .stream_dataset import build_selection, stream
source = str(settings.get("stream_source") or "array").lower()
src = str(settings.get("src") or "")
merged = str(settings.get("merged_folder") or os.path.join(src, "merged"))
database = str(settings.get("database")
or os.path.join(src, "measurements", "measurements.db"))
object_type = str(settings.get("object_array") or "cell")
table = str(settings.get("table") or "")
if not table:
try:
table = reserve_png_table(database)
except Exception: # noqa: BLE001
LOG.warning("could not reserve a png_list table in %s", database,
exc_info=True)
return {"written": 0, "missing": 0, "fields": 0, "folders": [],
"table": "",
"trouble": [f"could not open {database} to reserve a "
f"table for this set"]}
destination = str(settings.get("dst")
or os.path.join(src, crops_folder_for(table)))
objects = None
if source == "database":
objects = read_objects_from_database(database, object_type)
if objects is None or not len(objects):
return {"written": 0, "missing": 0, "fields": 0, "folders": [],
"table": "",
"trouble": [f"no {object_type} rows in {database}"]}
selection, selection_path = build_selection(
destination, objects=objects, merged_folder=merged,
object_array=object_type,
test_split=float(settings.get("test_split") or 0.0),
seed=int(settings.get("random_seed") or 0))
selection = filter_selection(selection, settings)
if not len(selection):
return {"written": 0, "missing": 0, "fields": 0, "folders": [],
"table": "", "selection": selection_path,
"trouble": ["every object was filtered out; the reserved "
f"table {table} is empty"]}
written: List[str] = []
png_size = list(settings.get("png_size") or (224, 224))
channels = list(settings.get("channel_arrays") or (0, 1, 2))
def _write(path, array):
"""Write and register one streamed crop through Measure's writer.
:param path: proposed crop path; its extension is replaced with
``.png``.
:param array: crop array to normalize, select, pad, resize, and save.
:returns: None. The shared writer's returned path is appended to the
captured list using the captured channels and PNG size, preserving
alignment with the subsequently registered selection rows.
"""
from .measure import _save_object_crop
target = os.path.splitext(str(path))[0] + ".png"
written.append(_save_object_crop(array, channels, target, png_size))
report = stream(
selection, merged, destination,
channel_arrays=list(settings.get("channel_arrays") or (0, 1, 2)),
bounding_box=(True if source == "database"
else bool(settings.get("bounding_box", True))),
crop_mode=object_type, write=_write)
report["selection"] = selection_path
if written:
frame = png_list_frame(selection.head(len(written)), written)
report["table"] = write_png_list(database, frame, table=table)
else:
report["table"] = ""
report.setdefault("trouble", []).append(
f"nothing was written; the reserved table {table} is empty")
return report
[docs]
def read_objects_from_database(db_path: str, object_type: str
) -> Optional[pd.DataFrame]:
"""The object table a streamed set can be built from.
:param db_path: the measurements database.
:param object_type: ``cell``, ``nucleus``, ``pathogen`` …
:returns: the rows, or ``None`` when the table is not there.
"""
if not os.path.isfile(str(db_path)):
return None
try:
frame, = read_database(
db_path,
object_type,
canonicalise=False,
report=None,
migrate=False,
read_only=True,
)
except ValueError as error:
if str(error).startswith("Table not found in database:"):
return None
raise
return frame