Source code for itasc.contact_analysis.quantifier

"""The quantifier seam: per-position compute units, discovered by subclassing.

A :class:`Quantifier` turns a position's source files (:class:`PositionInputs`)
into a persisted, plottable quantity. Contacts is the first one; new quantities
(nucleus-track kinematics, nucleus-vs-cell offset, cell shape, …) drop in as
modules under :mod:`itasc.contact_analysis.quantifiers` without
touching the studio.

The mechanics mirror :mod:`itasc.napari.contact_analysis.plugins`: subclassing with a
non-empty ``quantity_id`` auto-registers the quantifier, and
:func:`available_quantifiers` imports every module in the ``quantifiers`` package
so its plugins self-register. This module stays backend-only (no Qt / napari) so
the standalone wheel and headless batch runs can use it.
"""
from __future__ import annotations

import importlib
import pkgutil
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any, ClassVar

__all__ = ["OUTPUT_SUBDIR", "PositionInputs", "Quantifier", "available_quantifiers"]

#: Per-position subfolder for quantifier artifacts that persist to disk. Numbered
#: to mirror the staged layout (``0_input`` … ``3_cell``). The dynamics
#: quantifiers land here (``cell_dynamics.h5``, ``nucleus_dynamics.h5``); the
#: cheap quantities (shape, relational, density, neighbor count) are pooled in
#: memory during ``run()`` rather than written to disk. Note ``contact_analysis.h5``
#: is *not* here — :class:`~itasc.contact_analysis.quantifiers.contacts.ContactsQuantifier`
#: overrides :meth:`Quantifier.default_output` to write it to the position root,
#: beside the committed ``cell_labels.tif`` / ``nucleus_labels.tif``.
OUTPUT_SUBDIR = "4_contact_analysis"




@dataclass(frozen=True)
class PositionInputs:
    """Resolved source files for one position. Quantifiers read what they need.

    Every field has a live consumer. A future track-based quantifier adds its own
    field (e.g. ``tracks_db_path``) in the same commit that first reads it — no
    speculative placeholders here.
    """

    position_dir: Path
    cell_labels_path: Path | None = None
    nucleus_labels_path: Path | None = None
    #: Physical pixel size (µm/px); ``None`` when unknown. Quantifiers that emit
    #: values in physical units (cell shape) require it.
    pixel_size_um: float | None = None
    #: Frame interval (seconds/frame); ``None`` when unknown. Quantifiers that
    #: emit time-derived values (track dynamics) require it.
    time_interval_s: float | None = None
    #: The position's built ``contact_analysis.h5``; ``None`` when contacts is not
    #: in the catalogue. The contacts-derived quantifiers (neighbor count /
    #: enrichment / z-score / density / signed contact length) read it as their input instead
    #: of re-running contact extraction.
    contact_analysis_path: Path | None = None



#: quantity_id -> quantifier class, populated by ``__init_subclass__``.
_REGISTRY: dict[str, type[Quantifier]] = {}




class Quantifier:
    """Base class for per-position quantifiers.

    Subclasses set ``quantity_id`` / ``display_name`` (and usually ``requires``)
    and implement :meth:`build` and :meth:`read`. A quantifier **owns its own
    persistence**: :meth:`build` writes whatever artifact format suits the
    quantity and :meth:`read` parses it back — the framework imposes no schema.
    """

    #: Stable key; an empty value marks an intermediate (non-registered) base.
    quantity_id: ClassVar[str] = ""
    #: Human-readable label shown wherever a quantity is selected.
    display_name: ClassVar[str] = ""
    #: ``PositionInputs`` field names this quantifier needs to build.
    requires: ClassVar[tuple[str, ...]] = ()
    #: The ``PositionInputs`` field this quantifier's artifact *populates*, if any
    #: (e.g. contacts populates ``contact_analysis_path``). A quantifier whose
    #: :attr:`requires` names another's :attr:`produces` is *derived from* it — the
    #: studio uses this to draw the build-dependency graph. Empty for a leaf
    #: quantity that only consumes raw source inputs.
    produces: ClassVar[str] = ""
    #: Default artifact file name (relative to a position); empty for an
    #: intermediate base that does not persist.
    default_output_name: ClassVar[str] = ""
    #: The index columns of this quantifier's ``object_table`` — its natural grain
    #: (e.g. ``("frame", "cell_id")``). A non-empty value marks the quantity as
    #: **aggregated**: :mod:`itasc.contact_analysis.shape_tables` pools it
    #: across positions into a table named by :attr:`quantity_id`, keyed on these
    #: columns. Empty ⇒ not aggregated into an index-keyed table (e.g. contacts).
    #: Value columns are namespaced by ``quantity_id`` so a later joined view across
    #: tables never has colliding names.
    table_keys: ClassVar[tuple[str, ...]] = ()
    #: Whether the studio threads the shared plot/build params (z-score shuffle
    #: count, density field-of-view) into :meth:`build` via ``params``. Off by
    #: default so a quantifier with its own ``params`` schema (contacts edge
    #: extraction, shape, dynamics) is never handed the shared bar's keys.
    wants_build_params: ClassVar[bool] = False
    #: Shared build-param keys that must be present and positive for this
    #: quantifier to build, mapped to the human label shown in the UI (e.g. cell
    #: density needs ``{"fov_area_mm2": "FOV area (mm²)"}``). The studio greys the
    #: metric out — rather than letting the build fail — until they are supplied.
    #: Empty for a quantifier that needs no shared param.
    required_build_params: ClassVar[dict[str, str]] = {}

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        if cls.quantity_id:
            _REGISTRY[cls.quantity_id] = cls



    def can_build(self, inputs: PositionInputs) -> bool:
        """True when *inputs* supplies every field named in :attr:`requires`."""
        return all(getattr(inputs, name, None) is not None for name in self.requires)




    def missing_build_params(self, params: Mapping[str, Any] | None) -> tuple[str, ...]:
        """Labels of :attr:`required_build_params` that *params* doesn't satisfy.

        A required key is satisfied when present and a positive number; anything
        else (absent, blank, ``None``, non-positive) counts as missing. The studio
        uses the returned labels to grey the metric out and explain why. Empty
        tuple ⇒ every required shared param is supplied."""
        params = params or {}
        missing: list[str] = []
        for key, label in self.required_build_params.items():
            value = params.get(key)
            if not (isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0):
                missing.append(label)
        return tuple(missing)




    def default_output(self, inputs: PositionInputs) -> Path:
        """Where this quantifier's artifact lives for *inputs*, by default.

        ``position_dir / OUTPUT_SUBDIR / default_output_name`` — every quantity
        lands in the shared :data:`OUTPUT_SUBDIR` folder, so each subclass sets
        just a bare file name. The studio uses this to decide a build's
        destination, so a second quantifier no longer inherits the contacts
        artifact path. Subclasses may override for richer layouts.
        """
        if not self.default_output_name:
            raise NotImplementedError(
                f"{type(self).__name__} sets no default_output_name"
            )
        return inputs.position_dir / OUTPUT_SUBDIR / self.default_output_name




    def is_built(self, output_path: Path) -> bool:
        """True when the artifact at *output_path* already exists."""
        return Path(output_path).is_file()




    def object_table(self, output_path: Path) -> Mapping[str, Any] | None:
        """A tidy, column-major per-object table for the plotting backend.

        At least a ``frame`` key plus a per-object key (e.g. ``cell_id``).
        Returns ``None`` when this quantifier produces no per-object table; the
        plotting backend then skips it.
        """
        return None




    def compute_object_table(
        self, inputs: PositionInputs, *, params: dict | None = None
    ) -> Mapping[str, Any] | None:
        """The pooled tidy table for one position, computed directly from *inputs*.

        A **pooled** quantifier (one that declares ``table_keys``) implements this:
        it returns the same column-major table that ``object_table`` used to return
        after a disk round-trip, but never touches disk. ``None`` when this position
        yields no rows. Producers (``contacts``) are not pooled and do not implement
        it. *params* carries the shared build knobs (e.g. ``fov_area_mm2``) for the
        quantifiers that opt in; per-position values (pixel size, frame interval)
        arrive via *inputs*.
        """
        raise NotImplementedError




    def build(
        self,
        inputs: PositionInputs,
        output_path: Path,
        *,
        params: dict | None = None,
        progress_cb: Callable[[int, int, str], None] | None = None,
    ) -> Path:  # pragma: no cover - overridden
        raise NotImplementedError




    def read(self, output_path: Path) -> Any:  # pragma: no cover - overridden
        raise NotImplementedError




def _import_quantifier_modules() -> None:
    """Import every (non-private) ``quantifiers`` submodule so it self-registers."""
    from . import quantifiers

    for info in pkgutil.iter_modules(quantifiers.__path__):
        if info.name.startswith("_"):
            continue
        importlib.import_module(f"{quantifiers.__name__}.{info.name}")




def available_quantifiers() -> list[type[Quantifier]]:
    """Return registered quantifier classes, sorted by display name."""
    _import_quantifier_modules()
    return sorted(_REGISTRY.values(), key=lambda cls: cls.display_name.lower())