Finished analysis

.gitignore

@@ -1,4 +1,8 @@
 .DS_Store
 __pycache__
-out/
+out*/
 data/test
+results*
+test_results
+data/dab_quantification_test
+full_data

ihc_qc.py

@@ -0,0 +1,246 @@
+"""QC visualization for spheroid segmentation and DAB quantification."""
+
+from pathlib import Path
+
+import cv2
+import numpy as np
+
+
+DEFAULT_QC_MAX_DIMENSION = 1200
+DEFAULT_DAB_VMAX = 0.3
+
+
+def _resize(
+    image: np.ndarray,
+    max_dimension: int,
+    *,
+    nearest: bool = False,
+) -> np.ndarray:
+    height, width = image.shape[:2]
+    scale = min(1.0, max_dimension / max(height, width))
+    if scale == 1:
+        return image.copy()
+
+    size = (round(width * scale), round(height * scale))
+    interpolation = cv2.INTER_NEAREST if nearest else cv2.INTER_AREA
+    return cv2.resize(image, size, interpolation=interpolation)
+
+
+def _draw_spheroids(
+    image: np.ndarray,
+    labels: np.ndarray,
+    boundary_spheroid_ids: set[int],
+) -> np.ndarray:
+    output = image.copy()
+
+    for spheroid_id in np.unique(labels):
+        if spheroid_id == 0:
+            continue
+
+        spheroid = labels == spheroid_id
+        contours, _ = cv2.findContours(
+            spheroid.astype(np.uint8),
+            cv2.RETR_EXTERNAL,
+            cv2.CHAIN_APPROX_SIMPLE,
+        )
+        outline_color = (
+            (255, 165, 0)
+            if spheroid_id in boundary_spheroid_ids
+            else (0, 255, 255)
+        )
+        cv2.drawContours(output, contours, -1, outline_color, 3)
+
+        y, x = np.nonzero(spheroid)
+        center = (round(x.mean()), round(y.mean()))
+        text = str(spheroid_id)
+        cv2.circle(output, center, 18, (0, 0, 0), cv2.FILLED)
+        cv2.putText(
+            output,
+            text,
+            (center[0] - 7 * len(text), center[1] + 7),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.65,
+            (255, 255, 255),
+            2,
+            cv2.LINE_AA,
+        )
+
+    return output
+
+
+def _draw_debris_outlines(
+    image: np.ndarray,
+    debris_mask: np.ndarray,
+) -> np.ndarray:
+    output = image.copy()
+    contours, _ = cv2.findContours(
+        debris_mask.astype(np.uint8),
+        cv2.RETR_EXTERNAL,
+        cv2.CHAIN_APPROX_SIMPLE,
+    )
+    cv2.drawContours(output, contours, -1, (255, 0, 255), 1)
+    return output
+
+
+def _add_title(image: np.ndarray, title: str) -> np.ndarray:
+    titled = cv2.copyMakeBorder(
+        image,
+        54,
+        0,
+        0,
+        0,
+        cv2.BORDER_CONSTANT,
+        value=(28, 28, 28),
+    )
+    cv2.putText(
+        titled,
+        title,
+        (18, 36),
+        cv2.FONT_HERSHEY_SIMPLEX,
+        0.8,
+        (255, 255, 255),
+        2,
+        cv2.LINE_AA,
+    )
+    return titled
+
+
+def create_qc_image(
+    preview: np.ndarray,
+    labels: np.ndarray,
+    debris_mask: np.ndarray,
+    dab: np.ndarray,
+    *,
+    boundary_spheroid_ids: set[int] | None = None,
+    positive_threshold: float | None = None,
+    dab_vmax: float = DEFAULT_DAB_VMAX,
+    max_dimension: int = DEFAULT_QC_MAX_DIMENSION,
+) -> np.ndarray:
+    """Return an RGB QC image with segmentation, debris, and DAB panels."""
+    if preview.shape[:2] != labels.shape:
+        raise ValueError("preview and labels must have the same height and width.")
+    if labels.shape != debris_mask.shape or labels.shape != dab.shape:
+        raise ValueError("labels, debris_mask, and dab must have the same shape.")
+    if max_dimension <= 0:
+        raise ValueError("max_dimension must be positive.")
+    if positive_threshold is not None and not np.isfinite(positive_threshold):
+        raise ValueError("positive_threshold must be finite.")
+    if not np.isfinite(dab_vmax) or dab_vmax <= 0:
+        raise ValueError("dab_vmax must be finite and positive.")
+
+    preview_small = _resize(preview, max_dimension)
+    labels_small = _resize(labels, max_dimension, nearest=True)
+    debris_small = _resize(
+        (debris_mask > 0).astype(np.uint8),
+        max_dimension,
+        nearest=True,
+    ).astype(bool)
+    dab_small = _resize(dab, max_dimension)
+    if boundary_spheroid_ids is None:
+        boundary_spheroid_ids = {
+            int(spheroid_id)
+            for spheroid_id in np.unique(
+                np.concatenate(
+                    (
+                        labels[0, :],
+                        labels[-1, :],
+                        labels[:, 0],
+                        labels[:, -1],
+                    )
+                )
+            )
+            if spheroid_id != 0
+        }
+
+    debris_outline = _draw_debris_outlines(preview_small, debris_small)
+    debris_outline = _draw_spheroids(
+        debris_outline,
+        labels_small,
+        boundary_spheroid_ids,
+    )
+    debris_outline = _add_title(
+        debris_outline,
+        "cyan: within edge tolerance | orange: exceeds tolerance | magenta: excluded",
+    )
+
+    valid = (labels_small > 0) & ~debris_small
+    heatmap = np.full((*dab_small.shape, 3), 235, dtype=np.uint8)
+    if valid.any():
+        scaled_dab = np.clip(dab_small / dab_vmax, 0, 1)
+        colors = cv2.applyColorMap(
+            (scaled_dab * 255).astype(np.uint8),
+            cv2.COLORMAP_INFERNO,
+        )
+        colors = cv2.cvtColor(colors, cv2.COLOR_BGR2RGB)
+        heatmap[valid] = colors[valid]
+
+    heatmap[debris_small & (labels_small > 0)] = (255, 0, 255)
+    heatmap = _draw_spheroids(
+        heatmap,
+        labels_small,
+        boundary_spheroid_ids,
+    )
+    heatmap = _add_title(
+        heatmap,
+        (
+            f"DAB signal (fixed 0-{dab_vmax:g}) | "
+            "orange: exceeds edge tolerance"
+        ),
+    )
+
+    panels = [debris_outline, heatmap]
+    if positive_threshold is not None:
+        positive = (
+            (labels > 0)
+            & (debris_mask == 0)
+            & (dab >= positive_threshold)
+        )
+        positive_small = _resize(
+            positive.astype(np.uint8),
+            max_dimension,
+            nearest=True,
+        ).astype(bool)
+
+        binary_positive = np.zeros((*labels_small.shape, 3), dtype=np.uint8)
+        binary_positive[positive_small] = (255, 255, 255)
+        binary_positive = _add_title(
+            binary_positive,
+            f"white: DAB-positive | threshold >= {positive_threshold:.4g}",
+        )
+        panels.append(binary_positive)
+
+    return np.concatenate(panels, axis=1)
+
+
+def save_qc_image(
+    output_path: str | Path,
+    preview: np.ndarray,
+    labels: np.ndarray,
+    debris_mask: np.ndarray,
+    dab: np.ndarray,
+    *,
+    boundary_spheroid_ids: set[int] | None = None,
+    positive_threshold: float | None = None,
+    dab_vmax: float = DEFAULT_DAB_VMAX,
+    max_dimension: int = DEFAULT_QC_MAX_DIMENSION,
+) -> Path:
+    """Create and save the RGB QC image, returning its output path."""
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    qc_image = create_qc_image(
+        preview,
+        labels,
+        debris_mask,
+        dab,
+        boundary_spheroid_ids=boundary_spheroid_ids,
+        positive_threshold=positive_threshold,
+        dab_vmax=dab_vmax,
+        max_dimension=max_dimension,
+    )
+    saved = cv2.imwrite(
+        str(output_path),
+        cv2.cvtColor(qc_image, cv2.COLOR_RGB2BGR),
+    )
+    if not saved:
+        raise OSError(f"Could not write QC image: {output_path}")
+    return output_path

ihc_quantification_gpt.py

@@ -0,0 +1,853 @@
+"""Quantify DAB staining per spheroid while excluding debris.
+
+The source image is never inpainted.  Spheroids and debris are detected on an
+8-bit preview, while H-DAB stain separation is performed on the original image
+values.  Results from multiple runs are merged into one wide CSV: each
+spheroid is a column and each measurement is a row.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import os
+import re
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+import cv2
+import numpy as np
+import tifffile
+from skimage.color import hdx_from_rgb, separate_stains
+from skimage.morphology import remove_small_holes
+from skimage.util import img_as_float32
+
+from cleaning import _as_rgb_uint8, create_debris_mask
+from ihc_qc import DEFAULT_QC_MAX_DIMENSION, save_qc_image
+
+
+IMAGE_EXTENSIONS = {".bmp", ".jpeg", ".jpg", ".png", ".tif", ".tiff"}
+DEFAULT_OUTPUT_DIR = Path("out/quantification")
+DEFAULT_ADAPTIVE_BLOCK_FRACTION = 0.14
+DEFAULT_ADAPTIVE_OFFSET = 15.0
+DEFAULT_OPENING_RADIUS = 2
+DEFAULT_CLOSING_RADIUS = 25
+DEFAULT_MIN_SPHEROID_AREA_FRACTION = 0.0025
+DEFAULT_MAX_HOLE_AREA_FRACTION = 0.00025
+CONCENTRATION_PATTERN = re.compile(
+    r"^(?P<value>\d+(?:[.,]\d+)?)\s*(?P<unit>pM|nM|uM|µM|μM|mM)$",
+    re.IGNORECASE,
+)
+MAGNIFICATION_PATTERN = re.compile(r"^\d+(?:[.,]\d+)?x$", re.IGNORECASE)
+CSV_METRICS = (
+    "image_name",
+    "source_path",
+    "cell_line",
+    "culture_model",
+    "treatment",
+    "concentration_value",
+    "concentration_unit",
+    "antibody",
+    "magnification",
+    "image_number",
+    "spheroid_id",
+    "touches_image_border",
+    "centroid_x_px",
+    "centroid_y_px",
+    "total_spheroid_area_px",
+    "segmented_tissue_area_px",
+    "valid_measured_area_px",
+    "debris_excluded_area_px",
+    "debris_excluded_fraction",
+    "mean_dab_intensity",
+    "median_dab_intensity",
+    "integrated_dab_intensity",
+    "positive_dab_fraction",
+    "dab_positive_threshold",
+    "stain_separation",
+    "background_correction",
+)
+
+
+@dataclass(frozen=True)
+class FilenameMetadata:
+    cell_line: str = ""
+    culture_model: str = ""
+    treatment: str = ""
+    concentration_value: float | None = None
+    concentration_unit: str = ""
+    antibody: str = ""
+    magnification: str = ""
+    image_number: int | None = None
+
+
+@dataclass(frozen=True)
+class SpheroidMeasurement:
+    spheroid_id: int
+    touches_border: bool
+    centroid_x: float
+    centroid_y: float
+    envelope_area: int
+    tissue_area: int
+    valid_area: int
+    debris_area: int
+    debris_fraction: float
+    mean_dab: float
+    median_dab: float
+    integrated_dab: float
+    positive_fraction: float | None
+
+
+@dataclass(frozen=True)
+class ImageResult:
+    image_path: Path
+    image_id: str
+    metadata: FilenameMetadata
+    labels: np.ndarray
+    debris_mask: np.ndarray
+    dab: np.ndarray
+    measurements: tuple[SpheroidMeasurement, ...]
+    dab_positive_threshold: float | None
+    background_corrected: bool
+
+
+def parse_filename(image_path: str | Path) -> FilenameMetadata:
+    """Extract experimental metadata from a conventionally named image."""
+    tokens = Path(image_path).stem.split()
+    magnification_index = next(
+        (
+            index
+            for index, token in enumerate(tokens)
+            if MAGNIFICATION_PATTERN.fullmatch(token)
+        ),
+        None,
+    )
+
+    if magnification_index is None or magnification_index < 3:
+        return FilenameMetadata()
+
+    number_index = magnification_index + 1
+    if number_index != len(tokens) - 1 or not tokens[number_index].isdigit():
+        return FilenameMetadata()
+
+    experiment_tokens = tokens[2 : magnification_index - 1]
+    concentration_value = None
+    concentration_unit = ""
+    treatment_tokens: list[str] = []
+
+    for token in experiment_tokens:
+        concentration_match = CONCENTRATION_PATTERN.fullmatch(token)
+        if concentration_match and concentration_value is None:
+            concentration_value = float(
+                concentration_match.group("value").replace(",", ".")
+            )
+            unit = concentration_match.group("unit").replace("µ", "u").replace(
+                "μ", "u"
+            )
+            concentration_unit = {
+                "pm": "pM",
+                "nm": "nM",
+                "um": "uM",
+                "mm": "mM",
+            }[unit.lower()]
+        else:
+            treatment_tokens.append(token)
+
+    return FilenameMetadata(
+        cell_line=tokens[0],
+        culture_model=tokens[1],
+        treatment=" ".join(treatment_tokens),
+        concentration_value=concentration_value,
+        concentration_unit=concentration_unit,
+        antibody=tokens[magnification_index - 1],
+        magnification=tokens[magnification_index],
+        image_number=int(tokens[number_index]),
+    )
+
+
+def _select_rgb_plane(image: np.ndarray) -> np.ndarray:
+    """Select a single RGB plane without changing the source bit depth."""
+    image = np.asarray(image)
+
+    while image.ndim > 2 and 1 in image.shape:
+        image = np.squeeze(image)
+
+    if image.ndim != 3:
+        raise ValueError("Quantification requires a 2D RGB image.")
+
+    if image.shape[-1] in {3, 4}:
+        image = image[..., :3]
+    elif image.shape[0] in {3, 4}:
+        image = np.moveaxis(image[:3], 0, -1)
+    else:
+        raise ValueError("Quantification requires an RGB image with 3 channels.")
+
+    return image
+
+
+def read_original_rgb(image_path: str | Path) -> np.ndarray:
+    """Read an RGB image while preserving its original numeric values."""
+    image_path = Path(image_path)
+
+    if image_path.suffix.lower() in {".tif", ".tiff"}:
+        return _select_rgb_plane(tifffile.imread(image_path))
+
+    image = cv2.imread(str(image_path), cv2.IMREAD_UNCHANGED)
+    if image is None:
+        raise FileNotFoundError(f"Could not read image: {image_path}")
+
+    image = _select_rgb_plane(image)
+    return cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+
+
+def _odd_block_size(shape: tuple[int, int], requested: int | None) -> int:
+    shortest_side = min(shape)
+    if shortest_side < 3:
+        raise ValueError("Image is too small for adaptive segmentation.")
+
+    if requested is None:
+        block_size = round(shortest_side * DEFAULT_ADAPTIVE_BLOCK_FRACTION)
+    else:
+        block_size = requested
+
+    block_size = max(3, min(block_size, shortest_side))
+    if block_size % 2 == 0:
+        block_size -= 1
+    return block_size
+
+
+def segment_spheroids(
+    preview_rgb: np.ndarray,
+    *,
+    adaptive_block_size: int | None = None,
+    adaptive_offset: float = DEFAULT_ADAPTIVE_OFFSET,
+    opening_radius: int = DEFAULT_OPENING_RADIUS,
+    closing_radius: int = DEFAULT_CLOSING_RADIUS,
+    min_area_fraction: float = DEFAULT_MIN_SPHEROID_AREA_FRACTION,
+    max_hole_area_fraction: float = DEFAULT_MAX_HOLE_AREA_FRACTION,
+) -> np.ndarray:
+    """Return row-major spheroid labels using local contrast segmentation."""
+    if not 0 < min_area_fraction < 1:
+        raise ValueError("min_area_fraction must be between 0 and 1.")
+    if opening_radius < 0 or closing_radius < 0:
+        raise ValueError("Morphology radii must be non-negative.")
+    if not 0 <= max_hole_area_fraction < 1:
+        raise ValueError("max_hole_area_fraction must be between 0 and 1.")
+
+    gray = cv2.cvtColor(preview_rgb, cv2.COLOR_RGB2GRAY)
+    block_size = _odd_block_size(gray.shape, adaptive_block_size)
+    tissue_seed = cv2.adaptiveThreshold(
+        gray,
+        255,
+        cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
+        cv2.THRESH_BINARY_INV,
+        block_size,
+        adaptive_offset,
+    )
+
+    # Remove isolated dark specks before closing so they cannot form bridges
+    # between a spheroid and unrelated debris or vignetted image edges.
+    if opening_radius:
+        kernel_size = 2 * opening_radius + 1
+        opening_kernel = cv2.getStructuringElement(
+            cv2.MORPH_ELLIPSE,
+            (kernel_size, kernel_size),
+        )
+        tissue_seed = cv2.morphologyEx(
+            tissue_seed,
+            cv2.MORPH_OPEN,
+            opening_kernel,
+        )
+
+    if closing_radius:
+        kernel_size = 2 * closing_radius + 1
+        closing_kernel = cv2.getStructuringElement(
+            cv2.MORPH_ELLIPSE,
+            (kernel_size, kernel_size),
+        )
+        tissue_seed = cv2.morphologyEx(
+            tissue_seed,
+            cv2.MORPH_CLOSE,
+            closing_kernel,
+        )
+
+    max_hole_area = round(
+        preview_rgb.shape[0]
+        * preview_rgb.shape[1]
+        * max_hole_area_fraction
+    )
+    if max_hole_area:
+        tissue_seed = remove_small_holes(
+            tissue_seed.astype(bool),
+            max_size=max_hole_area,
+            connectivity=2,
+        ).astype(np.uint8)
+
+    count, initial_labels, stats, centroids = cv2.connectedComponentsWithStats(
+        tissue_seed,
+        connectivity=8,
+    )
+    min_area = round(preview_rgb.shape[0] * preview_rgb.shape[1] * min_area_fraction)
+    components = [
+        component
+        for component in range(1, count)
+        if stats[component, cv2.CC_STAT_AREA] >= min_area
+    ]
+    components.sort(
+        key=lambda component: (
+            centroids[component, 1],
+            centroids[component, 0],
+        )
+    )
+
+    labels = np.zeros(initial_labels.shape, dtype=np.int32)
+    for spheroid_id, component in enumerate(components, start=1):
+        labels[initial_labels == component] = spheroid_id
+
+    return labels
+
+
+def _quadratic_design(x: np.ndarray, y: np.ndarray) -> np.ndarray:
+    return np.column_stack((np.ones_like(x), x, y, x * x, x * y, y * y))
+
+
+def spheroid_envelope(labels: np.ndarray) -> np.ndarray:
+    """Fill each spheroid's external contour for size and background fitting."""
+    envelope = np.zeros(labels.shape, dtype=np.uint8)
+    for spheroid_id in range(1, int(labels.max()) + 1):
+        component = (labels == spheroid_id).astype(np.uint8)
+        contours, _ = cv2.findContours(
+            component,
+            cv2.RETR_EXTERNAL,
+            cv2.CHAIN_APPROX_SIMPLE,
+        )
+        cv2.drawContours(envelope, contours, -1, spheroid_id, cv2.FILLED)
+    return envelope
+
+
+def _fit_background_coefficients(
+    rgb: np.ndarray,
+    background_mask: np.ndarray,
+    *,
+    target_samples: int = 100_000,
+) -> np.ndarray | None:
+    """Fit a robust quadratic illumination surface to empty background."""
+    height, width = background_mask.shape
+    stride = max(1, int(np.sqrt(background_mask.size / target_samples)))
+    sampled_mask = background_mask[::stride, ::stride]
+    sampled_y, sampled_x = np.nonzero(sampled_mask)
+    if sampled_x.size < 1_000:
+        return None
+
+    sampled_y = sampled_y * stride
+    sampled_x = sampled_x * stride
+    x = sampled_x.astype(np.float64) / max(width - 1, 1) * 2 - 1
+    y = sampled_y.astype(np.float64) / max(height - 1, 1) * 2 - 1
+    design = _quadratic_design(x, y)
+
+    coefficients = np.zeros((3, design.shape[1]), dtype=np.float64)
+    for channel in range(3):
+        values = rgb[sampled_y, sampled_x, channel].astype(np.float64)
+        keep = np.ones(values.shape, dtype=bool)
+
+        for _ in range(4):
+            coefficients[channel], *_ = np.linalg.lstsq(
+                design[keep],
+                values[keep],
+                rcond=None,
+            )
+            residuals = values - design @ coefficients[channel]
+            center = np.median(residuals[keep])
+            mad = np.median(np.abs(residuals[keep] - center))
+            if mad <= np.finfo(np.float64).eps:
+                break
+            robust_sigma = 1.4826 * mad
+            new_keep = (
+                (residuals >= center - 2.5 * robust_sigma)
+                & (residuals <= center + 3.5 * robust_sigma)
+            )
+            if new_keep.sum() < 1_000 or np.array_equal(new_keep, keep):
+                break
+            keep = new_keep
+
+    return coefficients
+
+
+def correct_background_illumination(
+    rgb: np.ndarray,
+    tissue_mask: np.ndarray,
+    debris_mask: np.ndarray,
+) -> tuple[np.ndarray, bool]:
+    """Normalize smooth per-channel illumination using empty background."""
+    background_mask = ~tissue_mask & ~debris_mask
+    coefficients = _fit_background_coefficients(rgb, background_mask)
+    if coefficients is None:
+        return rgb, False
+
+    height, width = tissue_mask.shape
+    x = np.linspace(-1, 1, width, dtype=np.float32)[None, :]
+    y = np.linspace(-1, 1, height, dtype=np.float32)[:, None]
+    corrected = rgb.copy()
+
+    for channel in range(3):
+        c0, cx, cy, cxx, cxy, cyy = coefficients[channel]
+        surface = (
+            c0
+            + cx * x
+            + cy * y
+            + cxx * x * x
+            + cxy * x * y
+            + cyy * y * y
+        ).astype(np.float32)
+        surface = np.clip(surface, 0.05, 1.5)
+        corrected[..., channel] = np.clip(
+            corrected[..., channel] / surface,
+            0,
+            1,
+        )
+
+    return corrected, True
+
+
+def extract_dab(rgb: np.ndarray) -> np.ndarray:
+    """Extract the DAB component with scikit-image's fixed H-DAB matrix."""
+    stains = separate_stains(rgb, hdx_from_rgb)
+    dab = stains[..., 1].astype(np.float32)
+    return dab
+
+
+def _touches_border(component_mask: np.ndarray) -> bool:
+    return bool(
+        component_mask[0].any()
+        or component_mask[-1].any()
+        or component_mask[:, 0].any()
+        or component_mask[:, -1].any()
+    )
+
+
+def measure_spheroids(
+    labels: np.ndarray,
+    debris_mask: np.ndarray,
+    dab: np.ndarray,
+    dab_positive_threshold: float | None,
+) -> tuple[SpheroidMeasurement, ...]:
+    measurements: list[SpheroidMeasurement] = []
+    debris = debris_mask.astype(bool)
+    envelope_labels = spheroid_envelope(labels)
+
+    for spheroid_id in range(1, int(labels.max()) + 1):
+        spheroid = labels == spheroid_id
+        envelope = envelope_labels == spheroid_id
+        valid = spheroid & ~debris
+        tissue_area = int(spheroid.sum())
+        envelope_area = int(envelope.sum())
+        valid_area = int(valid.sum())
+        debris_area = tissue_area - valid_area
+        if not valid_area:
+            continue
+
+        y, x = np.nonzero(spheroid)
+        dab_values = dab[valid]
+        positive_fraction = None
+        if dab_positive_threshold is not None:
+            positive_fraction = float(
+                np.count_nonzero(dab_values >= dab_positive_threshold) / valid_area
+            )
+
+        measurements.append(
+            SpheroidMeasurement(
+                spheroid_id=spheroid_id,
+                touches_border=_touches_border(spheroid),
+                centroid_x=float(x.mean()),
+                centroid_y=float(y.mean()),
+                envelope_area=envelope_area,
+                tissue_area=tissue_area,
+                valid_area=valid_area,
+                debris_area=debris_area,
+                debris_fraction=debris_area / tissue_area,
+                mean_dab=float(dab_values.mean()),
+                median_dab=float(np.median(dab_values)),
+                integrated_dab=float(dab_values.sum(dtype=np.float64)),
+                positive_fraction=positive_fraction,
+            )
+        )
+
+    return tuple(measurements)
+
+
+def _safe_identifier(value: str) -> str:
+    identifier = re.sub(r"[^A-Za-z0-9._ -]+", "_", value).strip(" ._")
+    return identifier or "image"
+
+
+def quantify_image(
+    image_path: str | Path,
+    *,
+    dab_positive_threshold: float | None = None,
+    adaptive_block_size: int | None = None,
+    adaptive_offset: float = DEFAULT_ADAPTIVE_OFFSET,
+    opening_radius: int = DEFAULT_OPENING_RADIUS,
+    closing_radius: int = DEFAULT_CLOSING_RADIUS,
+    min_area_fraction: float = DEFAULT_MIN_SPHEROID_AREA_FRACTION,
+    max_hole_area_fraction: float = DEFAULT_MAX_HOLE_AREA_FRACTION,
+    background_correction: bool = True,
+) -> tuple[ImageResult, np.ndarray]:
+    image_path = Path(image_path)
+    original = read_original_rgb(image_path)
+    preview = _as_rgb_uint8(original)
+    labels = segment_spheroids(
+        preview,
+        adaptive_block_size=adaptive_block_size,
+        adaptive_offset=adaptive_offset,
+        opening_radius=opening_radius,
+        closing_radius=closing_radius,
+        min_area_fraction=min_area_fraction,
+        max_hole_area_fraction=max_hole_area_fraction,
+    )
+    if not labels.max():
+        raise ValueError(
+            "No spheroids were detected. Review the QC segmentation settings."
+        )
+
+    debris_mask = create_debris_mask(preview).astype(bool)
+    rgb = img_as_float32(original)
+    was_corrected = False
+    if background_correction:
+        rgb, was_corrected = correct_background_illumination(
+            rgb,
+            spheroid_envelope(labels) > 0,
+            debris_mask,
+        )
+
+    dab = extract_dab(rgb)
+    measurements = measure_spheroids(
+        labels,
+        debris_mask,
+        dab,
+        dab_positive_threshold,
+    )
+    if not measurements:
+        raise ValueError("No spheroid contained valid pixels after debris exclusion.")
+
+    result = ImageResult(
+        image_path=image_path.resolve(),
+        image_id=_safe_identifier(image_path.name),
+        metadata=parse_filename(image_path),
+        labels=labels,
+        debris_mask=debris_mask,
+        dab=dab,
+        measurements=measurements,
+        dab_positive_threshold=dab_positive_threshold,
+        background_corrected=was_corrected,
+    )
+    return result, preview
+
+
+def _format_float(value: float | None) -> str:
+    if value is None:
+        return ""
+    return f"{value:.8g}"
+
+
+def _measurement_values(
+    result: ImageResult,
+    measurement: SpheroidMeasurement,
+) -> dict[str, str]:
+    metadata = result.metadata
+    return {
+        "image_name": result.image_path.name,
+        "source_path": str(result.image_path),
+        "cell_line": metadata.cell_line,
+        "culture_model": metadata.culture_model,
+        "treatment": metadata.treatment,
+        "concentration_value": _format_float(metadata.concentration_value),
+        "concentration_unit": metadata.concentration_unit,
+        "antibody": metadata.antibody,
+        "magnification": metadata.magnification,
+        "image_number": (
+            str(metadata.image_number) if metadata.image_number is not None else ""
+        ),
+        "spheroid_id": str(measurement.spheroid_id),
+        "touches_image_border": str(measurement.touches_border).lower(),
+        "centroid_x_px": _format_float(measurement.centroid_x),
+        "centroid_y_px": _format_float(measurement.centroid_y),
+        "total_spheroid_area_px": str(measurement.envelope_area),
+        "segmented_tissue_area_px": str(measurement.tissue_area),
+        "valid_measured_area_px": str(measurement.valid_area),
+        "debris_excluded_area_px": str(measurement.debris_area),
+        "debris_excluded_fraction": _format_float(measurement.debris_fraction),
+        "mean_dab_intensity": _format_float(measurement.mean_dab),
+        "median_dab_intensity": _format_float(measurement.median_dab),
+        "integrated_dab_intensity": _format_float(measurement.integrated_dab),
+        "positive_dab_fraction": _format_float(measurement.positive_fraction),
+        "dab_positive_threshold": _format_float(result.dab_positive_threshold),
+        "stain_separation": "scikit-image H-DAB (hdx_from_rgb)",
+        "background_correction": (
+            "quadratic empty-background fit"
+            if result.background_corrected
+            else "none"
+        ),
+    }
+
+
+def _read_existing_results(
+    csv_path: Path,
+) -> tuple[list[str], dict[str, dict[str, str]], list[str]]:
+    if not csv_path.exists():
+        return [], {}, []
+
+    with csv_path.open(newline="", encoding="utf-8") as handle:
+        reader = csv.reader(handle)
+        rows = list(reader)
+
+    if not rows:
+        return [], {}, []
+    if not rows[0] or rows[0][0] != "metric":
+        raise ValueError(
+            f"Existing results file does not use the expected wide format: {csv_path}"
+        )
+
+    columns = rows[0][1:]
+    values: dict[str, dict[str, str]] = {}
+    metric_order: list[str] = []
+    for row in rows[1:]:
+        if not row:
+            continue
+        metric = row[0]
+        metric_order.append(metric)
+        padded = row[1:] + [""] * max(0, len(columns) - len(row[1:]))
+        values[metric] = dict(zip(columns, padded[: len(columns)], strict=True))
+
+    return columns, values, metric_order
+
+
+def update_results_csv(csv_path: str | Path, results: list[ImageResult]) -> None:
+    """Replace reprocessed images and append new spheroid columns atomically."""
+    csv_path = Path(csv_path)
+    csv_path.parent.mkdir(parents=True, exist_ok=True)
+    columns, values, previous_metric_order = _read_existing_results(csv_path)
+    source_paths = {str(result.image_path) for result in results}
+    source_row = values.get("source_path", {})
+    removed_columns = {
+        column for column in columns if source_row.get(column) in source_paths
+    }
+    columns = [column for column in columns if column not in removed_columns]
+    for metric_values in values.values():
+        for column in removed_columns:
+            metric_values.pop(column, None)
+
+    for result in results:
+        base_prefix = result.image_id
+        proposed_columns = [
+            f"{base_prefix}::spheroid_{measurement.spheroid_id:03d}"
+            for measurement in result.measurements
+        ]
+        if any(column in columns for column in proposed_columns):
+            digest = hashlib.sha1(
+                str(result.image_path).encode("utf-8")
+            ).hexdigest()[:8]
+            base_prefix = f"{base_prefix} [{digest}]"
+
+        for measurement in result.measurements:
+            column = f"{base_prefix}::spheroid_{measurement.spheroid_id:03d}"
+            columns.append(column)
+            for metric, value in _measurement_values(result, measurement).items():
+                values.setdefault(metric, {})[column] = value
+
+    metric_order = list(CSV_METRICS)
+    metric_order.extend(
+        metric
+        for metric in previous_metric_order
+        if metric not in metric_order
+    )
+    metric_order.extend(
+        metric for metric in values if metric not in metric_order
+    )
+
+    with tempfile.NamedTemporaryFile(
+        mode="w",
+        newline="",
+        encoding="utf-8",
+        dir=csv_path.parent,
+        prefix=f".{csv_path.name}.",
+        suffix=".tmp",
+        delete=False,
+    ) as handle:
+        temporary_path = Path(handle.name)
+        writer = csv.writer(handle)
+        writer.writerow(["metric", *columns])
+        for metric in metric_order:
+            metric_values = values.get(metric, {})
+            writer.writerow(
+                [metric, *(metric_values.get(column, "") for column in columns)]
+            )
+
+    os.replace(temporary_path, csv_path)
+
+
+def discover_images(inputs: list[str]) -> list[Path]:
+    images: list[Path] = []
+    seen: set[Path] = set()
+
+    for input_value in inputs:
+        input_path = Path(input_value)
+        if input_path.is_dir():
+            candidates = sorted(
+                path
+                for path in input_path.rglob("*")
+                if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS
+            )
+        elif input_path.is_file():
+            if input_path.suffix.lower() not in IMAGE_EXTENSIONS:
+                raise ValueError(f"Unsupported image extension: {input_path}")
+            candidates = [input_path]
+        else:
+            raise FileNotFoundError(f"Input does not exist: {input_path}")
+
+        for candidate in candidates:
+            resolved = candidate.resolve()
+            if resolved not in seen:
+                images.append(candidate)
+                seen.add(resolved)
+
+    if not images:
+        raise ValueError("No supported images were found.")
+    return images
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Quantify H-DAB signal per spheroid and update a wide results CSV."
+        )
+    )
+    parser.add_argument(
+        "images",
+        nargs="+",
+        help="Image files or directories. Directories are searched recursively.",
+    )
+    parser.add_argument(
+        "--output-dir",
+        type=Path,
+        default=DEFAULT_OUTPUT_DIR,
+        help=f"Results directory (default: {DEFAULT_OUTPUT_DIR}).",
+    )
+    parser.add_argument(
+        "--dab-positive-threshold",
+        type=float,
+        default=None,
+        help=(
+            "Fixed H-DAB threshold derived from negative controls. If omitted, "
+            "positive fractions are left blank."
+        ),
+    )
+    parser.add_argument(
+        "--adaptive-block-size",
+        type=int,
+        default=None,
+        help="Odd local-threshold window (default: 14%% of shortest image side).",
+    )
+    parser.add_argument(
+        "--adaptive-offset",
+        type=float,
+        default=DEFAULT_ADAPTIVE_OFFSET,
+        help=f"Local-threshold offset (default: {DEFAULT_ADAPTIVE_OFFSET:g}).",
+    )
+    parser.add_argument(
+        "--opening-radius",
+        type=int,
+        default=DEFAULT_OPENING_RADIUS,
+        help=f"Speck-removal radius in pixels (default: {DEFAULT_OPENING_RADIUS}).",
+    )
+    parser.add_argument(
+        "--closing-radius",
+        type=int,
+        default=DEFAULT_CLOSING_RADIUS,
+        help=(
+            "Spheroid-mask closing radius in pixels "
+            f"(default: {DEFAULT_CLOSING_RADIUS})."
+        ),
+    )
+    parser.add_argument(
+        "--min-spheroid-area-fraction",
+        type=float,
+        default=DEFAULT_MIN_SPHEROID_AREA_FRACTION,
+        help=(
+            "Minimum spheroid area as a fraction of image area "
+            f"(default: {DEFAULT_MIN_SPHEROID_AREA_FRACTION:g})."
+        ),
+    )
+    parser.add_argument(
+        "--max-hole-area-fraction",
+        type=float,
+        default=DEFAULT_MAX_HOLE_AREA_FRACTION,
+        help=(
+            "Fill internal mask holes up to this fraction of image area "
+            f"(default: {DEFAULT_MAX_HOLE_AREA_FRACTION:g})."
+        ),
+    )
+    parser.add_argument(
+        "--no-background-correction",
+        action="store_true",
+        help="Disable quadratic empty-background illumination correction.",
+    )
+    parser.add_argument(
+        "--qc-max-dimension",
+        type=int,
+        default=DEFAULT_QC_MAX_DIMENSION,
+        help=(
+            "Maximum width or height of each QC panel "
+            f"(default: {DEFAULT_QC_MAX_DIMENSION})."
+        ),
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    image_paths = discover_images(args.images)
+    results: list[ImageResult] = []
+    qc_dir = args.output_dir / "qc"
+
+    if args.dab_positive_threshold is None:
+        print(
+            "Note: positive_dab_fraction will be blank until a fixed "
+            "--dab-positive-threshold is supplied."
+        )
+
+    for image_path in image_paths:
+        result, preview = quantify_image(
+            image_path,
+            dab_positive_threshold=args.dab_positive_threshold,
+            adaptive_block_size=args.adaptive_block_size,
+            adaptive_offset=args.adaptive_offset,
+            opening_radius=args.opening_radius,
+            closing_radius=args.closing_radius,
+            min_area_fraction=args.min_spheroid_area_fraction,
+            max_hole_area_fraction=args.max_hole_area_fraction,
+            background_correction=not args.no_background_correction,
+        )
+        qc_path = qc_dir / f"{_safe_identifier(result.image_path.stem)}_qc.png"
+        save_qc_image(
+            qc_path,
+            preview,
+            result.labels,
+            result.debris_mask,
+            result.dab,
+            max_dimension=args.qc_max_dimension,
+        )
+        results.append(result)
+        print(
+            f"{image_path}: {len(result.measurements)} spheroids; "
+            f"QC: {qc_path}"
+        )
+
+    csv_path = args.output_dir / "ihc_quantification.csv"
+    update_results_csv(csv_path, results)
+    print(f"Updated results: {csv_path}")
+
+
+if __name__ == "__main__":
+    main()

ihc_quantification_simplified.py

@@ -0,0 +1,504 @@
+import argparse
+import re
+from pathlib import Path
+
+import cv2
+import numpy as np
+import pandas as pd
+import tifffile as tif
+from scipy.ndimage import binary_fill_holes
+from skimage.color import hdx_from_rgb, separate_stains
+from skimage.segmentation import watershed
+from skimage.util import img_as_float32, img_as_ubyte
+
+from cleaning import INPAINT_RADIUS, _as_rgb_uint8, create_debris_mask
+from ihc_qc import save_qc_image
+
+
+IMAGE_EXTENSIONS = {".tif", ".tiff"}
+DEFAULT_OUTPUT_PATH = Path("out/quantification/ihc_quantification.csv")
+MAX_BOUNDARY_CONTACT_RATIO = 0.3
+WATERSHED_MARKER_CORE_FRACTION = 0.55
+WATERSHED_MINIMUM_RELATIVE_REGION_AREA = 0.5
+CONCENTRATION_PATTERN = re.compile(
+    r"^\d+(?:[.,]\d+)?(?:pM|nM|uM|µM|μM|mM)$",
+    re.IGNORECASE,
+)
+MAGNIFICATION_PATTERN = re.compile(r"^\d+(?:[.,]\d+)?x$", re.IGNORECASE)
+MARKER_PATTERN = re.compile(r"^M\d+(?:[.,]\d+)?$", re.IGNORECASE)
+
+
+def parse_filename(image_path: str | Path) -> dict:
+    parts = Path(image_path).stem.split()
+    if len(parts) < 5:
+        raise ValueError(
+            f"Could not read metadata from filename: {Path(image_path).name}"
+        )
+
+    magnification_index = next(
+        (
+            index
+            for index in range(len(parts) - 1, 1, -1)
+            if MAGNIFICATION_PATTERN.fullmatch(parts[index])
+        ),
+        None,
+    )
+    if magnification_index is None or magnification_index < 3:
+        raise ValueError(
+            f"Could not read magnification from filename: {Path(image_path).name}"
+        )
+
+    trailing_tokens = parts[magnification_index + 1 :]
+    number_indices = [
+        index for index, token in enumerate(trailing_tokens) if token.isdigit()
+    ]
+    if len(number_indices) != 1:
+        raise ValueError(
+            f"Could not read a unique image number from filename: "
+            f"{Path(image_path).name}"
+        )
+
+    number_index = number_indices[0]
+    image_number = int(trailing_tokens[number_index])
+    image_note = " ".join(
+        token for index, token in enumerate(trailing_tokens) if index != number_index
+    )
+
+    metadata_end = magnification_index
+    if MARKER_PATTERN.fullmatch(parts[magnification_index - 1]):
+        metadata_end -= 1
+    metadata_tokens = parts[2:metadata_end]
+    treatment_tokens = [
+        token
+        for token in metadata_tokens
+        if (token == "+" or not token.startswith("+"))
+        and not CONCENTRATION_PATTERN.fullmatch(token)
+    ]
+    if not treatment_tokens:
+        raise ValueError(
+            f"Could not read treatment from filename: {Path(image_path).name}"
+        )
+    treatment = re.sub(r"\s*\+\s*", "+", " ".join(treatment_tokens))
+
+    return {
+        "image_name": Path(image_path).name,
+        "image_path": str(Path(image_path).resolve()),
+        "treatment": treatment,
+        "image_note": image_note,
+        "image_number": image_number,
+    }
+
+
+def read_original_rgb(image_path: str) -> np.ndarray:
+    image_path = Path(image_path)
+    if image_path.suffix.lower() in {".tif", ".tiff"}:
+        return tif.imread(image_path)
+
+
+def extract_dab(rgb: np.ndarray) -> np.ndarray:
+    stains = separate_stains(rgb, hdx_from_rgb)
+    dab = stains[..., 1].astype(np.float32)
+
+    return dab
+
+
+def split_touching_spheroids(
+    tissue_mask: np.ndarray,
+    min_area: int,
+    marker_core_fraction: float = WATERSHED_MARKER_CORE_FRACTION,
+    minimum_relative_region_area: float = WATERSHED_MINIMUM_RELATIVE_REGION_AREA,
+) -> np.ndarray:
+    if not 0 < marker_core_fraction <= 1:
+        raise ValueError(
+            "marker_core_fraction must be greater than 0 and at most 1."
+        )
+    if not 0 <= minimum_relative_region_area <= 1:
+        raise ValueError(
+            "minimum_relative_region_area must be between 0 and 1."
+        )
+
+    count, components, stats, _ = cv2.connectedComponentsWithStats(
+        tissue_mask.astype(np.uint8),
+        connectivity=8,
+    )
+    kept_components = [
+        component
+        for component in range(1, count)
+        if stats[component, cv2.CC_STAT_AREA] >= min_area
+    ]
+    filtered_mask = np.isin(components, kept_components)
+
+    distance = cv2.distanceTransform(
+        filtered_mask.astype(np.uint8),
+        cv2.DIST_L2,
+        5,
+    )
+    marker_mask = np.zeros(filtered_mask.shape, dtype=np.uint8)
+
+    for component in kept_components:
+        component_mask = components == component
+        maximum_distance = distance[component_mask].max()
+        marker_mask[
+            component_mask & (distance >= marker_core_fraction * maximum_distance)
+        ] = 1
+
+    _, markers = cv2.connectedComponents(marker_mask, connectivity=8)
+    candidate_labels = watershed(
+        -distance,
+        markers,
+        mask=filtered_mask,
+    ).astype(np.int32)
+
+    labels = np.zeros(candidate_labels.shape, dtype=np.int32)
+    next_label = 1
+
+    for component in kept_components:
+        component_mask = components == component
+        regions = [
+            region
+            for region in np.unique(candidate_labels[component_mask])
+            if region != 0
+        ]
+        region_areas = [
+            int((candidate_labels[component_mask] == region).sum())
+            for region in regions
+        ]
+
+        split_is_balanced = len(regions) > 1 and min(
+            region_areas
+        ) >= minimum_relative_region_area * max(region_areas)
+        if not split_is_balanced:
+            labels[component_mask] = next_label
+            next_label += 1
+            continue
+
+        for region in regions:
+            labels[component_mask & (candidate_labels == region)] = next_label
+            next_label += 1
+
+    return labels
+
+
+def segment_spheroids(
+    preview: np.ndarray,
+    marker_core_fraction: float = WATERSHED_MARKER_CORE_FRACTION,
+    minimum_relative_region_area: float = WATERSHED_MINIMUM_RELATIVE_REGION_AREA,
+) -> np.ndarray:
+    gray = cv2.cvtColor(preview, cv2.COLOR_RGB2GRAY)
+
+    # thresholding
+    C = 15
+    block_size = round(min(gray.shape) * 0.14)
+    block_size = block_size - 1 if block_size % 2 == 0 else block_size
+    gray = cv2.cvtColor(preview, cv2.COLOR_RGB2GRAY)
+    tissue_thresh = cv2.adaptiveThreshold(
+        gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, block_size, C
+    )
+
+    # filtering
+    opening_kernel = 5  # largest noise I want removed
+    opening_kernel = cv2.getStructuringElement(
+        cv2.MORPH_ELLIPSE,
+        (opening_kernel, opening_kernel),
+    )
+    tissue_thresh = cv2.morphologyEx(
+        tissue_thresh,
+        cv2.MORPH_OPEN,
+        opening_kernel,
+    )
+
+    closing_kernel = 51  # smallest element I want to connect
+    closing_kernel = cv2.getStructuringElement(
+        cv2.MORPH_ELLIPSE,
+        (closing_kernel, closing_kernel),
+    )
+    tissue_thresh = cv2.morphologyEx(
+        tissue_thresh,
+        cv2.MORPH_CLOSE,
+        closing_kernel,
+    )
+
+    tissue_filled = binary_fill_holes(tissue_thresh > 0).astype(np.uint8) * 255
+
+    min_area_fraction = (
+        0.0025  # component must occupy at least 0.25% of the complete image
+    )
+    min_area = round(preview.shape[0] * preview.shape[1] * min_area_fraction)
+    split_labels = split_touching_spheroids(
+        tissue_filled > 0,
+        min_area,
+        marker_core_fraction=marker_core_fraction,
+        minimum_relative_region_area=minimum_relative_region_area,
+    )
+
+    components = []
+    for component in np.unique(split_labels):
+        if component == 0:
+            continue
+
+        y, x = np.nonzero(split_labels == component)
+        if len(x) >= min_area:
+            components.append((component, y.mean(), x.mean()))
+
+    components.sort(key=lambda component: (component[1], component[2]))
+
+    labels = np.zeros(split_labels.shape, dtype=np.int32)
+    for spheroid_id, (component, _, _) in enumerate(components, start=1):
+        labels[split_labels == component] = spheroid_id
+
+    return labels
+
+
+def measure_spheroids(
+    labels: np.ndarray,
+    debris_mask: np.ndarray,
+    dab: np.ndarray,
+    positive_threshold: None | float = None,
+    max_boundary_contact_ratio: float = MAX_BOUNDARY_CONTACT_RATIO,
+) -> list[dict]:
+    if max_boundary_contact_ratio < 0:
+        raise ValueError("max_boundary_contact_ratio must be nonnegative.")
+
+    measurements = []
+    debris = debris_mask > 0
+    image_boundary = np.zeros(labels.shape, dtype=bool)
+    image_boundary[[0, -1], :] = True
+    image_boundary[:, [0, -1]] = True
+
+    for spheroid_id in np.unique(labels):
+        if spheroid_id == 0:
+            continue
+
+        spheroid = labels == spheroid_id
+        valid = spheroid & ~debris
+
+        spheroid_area = int(spheroid.sum())
+        valid_area = int(valid.sum())
+        debris_area = int((spheroid & debris).sum())
+        boundary_contact_px = int((spheroid & image_boundary).sum())
+        equivalent_diameter = float(2 * np.sqrt(spheroid_area / np.pi))
+        boundary_contact_ratio = boundary_contact_px / equivalent_diameter
+        touches_image_boundary = boundary_contact_px > 0
+        exceeds_boundary_tolerance = bool(
+            boundary_contact_ratio > max_boundary_contact_ratio
+        )
+
+        if valid_area == 0:
+            continue
+
+        dab_values = dab[valid]
+
+        result = {
+            "spheroid_id": int(spheroid_id),
+            "touches_image_boundary": touches_image_boundary,
+            "exceeds_boundary_tolerance": exceeds_boundary_tolerance,
+            "boundary_contact_ratio": boundary_contact_ratio,
+            "spheroid_area_px": spheroid_area,
+            "valid_area_px": valid_area,
+            "debris_area_px": debris_area,
+            "debris_fraction": debris_area / spheroid_area,
+            "mean_dab": dab_values.mean(),
+            "median_dab": np.median(dab_values),
+            "p75_dab": np.percentile(dab_values, 75),
+            "p90_dab": np.percentile(dab_values, 90),
+            "p95_dab": np.percentile(dab_values, 95),
+            "p99_dab": np.percentile(dab_values, 99),
+        }
+
+        if positive_threshold is not None:
+            positive = valid & (dab >= positive_threshold)
+            positive_values = dab[positive]
+
+            result["positive_fraction"] = positive.sum() / valid.sum()
+            result["positive_area_px"] = positive.sum()
+            result["positive_mean"] = (
+                positive_values.mean() if positive_values.size else np.nan
+            )
+
+        measurements.append(result)
+
+    return measurements
+
+
+def quantify_image(
+    image_path: str | Path,
+    qc_path: str | Path | None = None,
+    positive_threshold: float | None = None,
+    max_boundary_contact_ratio: float = MAX_BOUNDARY_CONTACT_RATIO,
+    marker_core_fraction: float = WATERSHED_MARKER_CORE_FRACTION,
+    minimum_relative_region_area: float = WATERSHED_MINIMUM_RELATIVE_REGION_AREA,
+) -> list[dict]:
+    metadata = parse_filename(image_path)
+    original = read_original_rgb(image_path)
+    processing_preview = _as_rgb_uint8(original)
+    qc_preview = img_as_ubyte(original)
+
+    debris_mask = create_debris_mask(qc_preview)
+    cleaned_preview = cv2.inpaint(
+        processing_preview,
+        debris_mask,
+        INPAINT_RADIUS,
+        cv2.INPAINT_TELEA,
+    )
+    spheroid_labels = segment_spheroids(
+        cleaned_preview,
+        marker_core_fraction=marker_core_fraction,
+        minimum_relative_region_area=minimum_relative_region_area,
+    )
+    if not spheroid_labels.max():
+        raise ValueError(
+            "No spheroids were detected. Review the QC segmentation settings."
+        )
+
+    float_img = img_as_float32(original)
+    dab = extract_dab(float_img)
+    measurements = measure_spheroids(
+        spheroid_labels,
+        debris_mask,
+        dab,
+        positive_threshold=positive_threshold,
+        max_boundary_contact_ratio=max_boundary_contact_ratio,
+    )
+
+    if qc_path is not None:
+        boundary_spheroid_ids = {
+            measurement["spheroid_id"]
+            for measurement in measurements
+            if measurement["exceeds_boundary_tolerance"]
+        }
+        save_qc_image(
+            qc_path,
+            qc_preview,
+            spheroid_labels,
+            debris_mask,
+            dab,
+            boundary_spheroid_ids=boundary_spheroid_ids,
+            positive_threshold=positive_threshold,
+        )
+
+    return [{**metadata, **measurement} for measurement in measurements]
+
+
+def find_images(data_path: str | Path) -> list[Path]:
+    data_path = Path(data_path)
+    if data_path.is_file():
+        if data_path.suffix.lower() not in IMAGE_EXTENSIONS:
+            raise ValueError(f"Input is not a TIFF image: {data_path}")
+        image_paths = [data_path]
+    elif data_path.is_dir():
+        image_paths = sorted(
+            path
+            for path in data_path.rglob("*")
+            if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS
+        )
+    else:
+        raise FileNotFoundError(f"Data path does not exist: {data_path}")
+
+    if not image_paths:
+        raise ValueError(f"No TIFF images found in: {data_path}")
+
+    return image_paths
+
+
+def quantify_data(
+    data_path: str | Path,
+    qc_dir: str | Path | None = None,
+    positive_threshold: float | None = None,
+    max_boundary_contact_ratio: float = MAX_BOUNDARY_CONTACT_RATIO,
+    marker_core_fraction: float = WATERSHED_MARKER_CORE_FRACTION,
+    minimum_relative_region_area: float = WATERSHED_MINIMUM_RELATIVE_REGION_AREA,
+) -> pd.DataFrame:
+    measurements = []
+    qc_dir = Path(qc_dir) if qc_dir is not None else None
+
+    for image_path in find_images(data_path):
+        qc_path = qc_dir / f"{image_path.stem}_qc.png" if qc_dir is not None else None
+        image_measurements = quantify_image(
+            image_path,
+            qc_path,
+            positive_threshold=positive_threshold,
+            max_boundary_contact_ratio=max_boundary_contact_ratio,
+            marker_core_fraction=marker_core_fraction,
+            minimum_relative_region_area=minimum_relative_region_area,
+        )
+        measurements.extend(image_measurements)
+        message = f"{image_path}: {len(image_measurements)} spheroids"
+        if qc_path is not None:
+            message += f"; QC: {qc_path}"
+        print(message)
+
+    if not measurements:
+        raise ValueError("No valid spheroid measurements were produced.")
+
+    return pd.DataFrame(measurements)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Quantify spheroid DAB intensity and save one row per spheroid."
+    )
+    parser.add_argument(
+        "data_path",
+        type=Path,
+        help="TIFF image or directory containing TIFF images.",
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=DEFAULT_OUTPUT_PATH,
+        help=f"Output CSV path (default: {DEFAULT_OUTPUT_PATH}).",
+    )
+    parser.add_argument(
+        "--positive-threshold",
+        type=float,
+        default=None,
+        help="Optional DAB-positive threshold used for measurements and QC.",
+    )
+    parser.add_argument(
+        "--max-boundary-contact-ratio",
+        type=float,
+        default=MAX_BOUNDARY_CONTACT_RATIO,
+        help=(
+            "Maximum tolerated boundary contact ratio before a spheroid is flagged "
+            f"(default: {MAX_BOUNDARY_CONTACT_RATIO})."
+        ),
+    )
+    parser.add_argument(
+        "--watershed-marker-core-fraction",
+        type=float,
+        default=WATERSHED_MARKER_CORE_FRACTION,
+        help=(
+            "Distance-transform fraction used to create watershed markers "
+            f"(default: {WATERSHED_MARKER_CORE_FRACTION})."
+        ),
+    )
+    parser.add_argument(
+        "--watershed-minimum-relative-region-area",
+        type=float,
+        default=WATERSHED_MINIMUM_RELATIVE_REGION_AREA,
+        help=(
+            "Smallest accepted watershed region relative to the largest region "
+            f"(default: {WATERSHED_MINIMUM_RELATIVE_REGION_AREA})."
+        ),
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    qc_dir = args.output.parent / "qc"
+    results = quantify_data(
+        args.data_path,
+        qc_dir,
+        positive_threshold=args.positive_threshold,
+        max_boundary_contact_ratio=args.max_boundary_contact_ratio,
+        marker_core_fraction=args.watershed_marker_core_fraction,
+        minimum_relative_region_area=args.watershed_minimum_relative_region_area,
+    )
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    results.to_csv(args.output, index=False)
+    print(f"Saved {len(results)} spheroids to {args.output}")
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -73,7 +73,9 @@ dependencies = [
     "notebook-shim==0.2.4",
     "numpy==2.4.6",
     "opencv-python==4.13.0.92",
+    "openpyxl>=3.1.5",
     "packaging==26.2",
+    "pandas==3.0.3",
     "pandocfilters==1.5.1",
     "parso==0.8.7",
     "pexpect==4.9.0",
@@ -99,6 +101,7 @@ dependencies = [
     "rpds-py==2026.5.1",
     "scikit-image==0.26.0",
     "scipy==1.17.1",
+    "seaborn>=0.13.2",
     "send2trash==2.1.0",
     "setuptools==82.0.1",
     "six==1.17.0",

requirements.txt

@@ -2,3 +2,5 @@ numpy==2.2.6
 opencv-python-headless==4.13.0.92
 imagecodecs==2026.6.26
 tifffile==2025.5.10
+scikit-image==0.26.0
+pandas==3.0.3

uv.lock

@@ -552,6 +552,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/07/6c/aa3f2f849e01cb6a001cd8554a88d4c77c5c1a31c95bdf1cf9301e6d9ef4/defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61", size = 25604, upload-time = "2021-03-08T10:59:24.45Z" },
 ]
 
+[[package]]
+name = "et-xmlfile"
+version = "2.0.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/d3/38/af70d7ab1ae9d4da450eeec1fa3918940a5fafb9055e934af8d6eb0c2313/et_xmlfile-2.0.0.tar.gz", hash = "sha256:dab3f4764309081ce75662649be815c4c9081e88f0837825f90fd28317d4da54", size = 17234, upload-time = "2024-10-25T17:25:40.039Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c1/8b/5fe2cc11fee489817272089c4203e679c63b570a5aaeb18d852ae3cbba6a/et_xmlfile-2.0.0-py3-none-any.whl", hash = "sha256:7a91720bc756843502c3b7504c77b8fe44217c85c537d85037f0f536151b2caa", size = 18059, upload-time = "2024-10-25T17:25:39.051Z" },
+]
+
 [[package]]
 name = "executing"
 version = "2.2.1"
@@ -905,7 +914,9 @@ dependencies = [
     { name = "notebook-shim" },
     { name = "numpy" },
     { name = "opencv-python" },
+    { name = "openpyxl" },
     { name = "packaging" },
+    { name = "pandas" },
     { name = "pandocfilters" },
     { name = "parso" },
     { name = "pexpect" },
@@ -931,6 +942,7 @@ dependencies = [
     { name = "rpds-py" },
     { name = "scikit-image" },
     { name = "scipy" },
+    { name = "seaborn" },
     { name = "send2trash" },
     { name = "setuptools" },
     { name = "six" },
@@ -1024,7 +1036,9 @@ requires-dist = [
     { name = "notebook-shim", specifier = "==0.2.4" },
     { name = "numpy", specifier = "==2.4.6" },
     { name = "opencv-python", specifier = "==4.13.0.92" },
+    { name = "openpyxl", specifier = ">=3.1.5" },
     { name = "packaging", specifier = "==26.2" },
+    { name = "pandas", specifier = "==3.0.3" },
     { name = "pandocfilters", specifier = "==1.5.1" },
     { name = "parso", specifier = "==0.8.7" },
     { name = "pexpect", specifier = "==4.9.0" },
@@ -1050,6 +1064,7 @@ requires-dist = [
     { name = "rpds-py", specifier = "==2026.5.1" },
     { name = "scikit-image", specifier = "==0.26.0" },
     { name = "scipy", specifier = "==1.17.1" },
+    { name = "seaborn", specifier = ">=0.13.2" },
     { name = "send2trash", specifier = "==2.1.0" },
     { name = "setuptools", specifier = "==82.0.1" },
     { name = "six", specifier = "==1.17.0" },
@@ -1925,6 +1940,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e9/a5/1be1516390333ff9be3a9cb648c9f33df79d5096e5884b5df71a588af463/opencv_python-4.13.0.92-cp37-abi3-win_amd64.whl", hash = "sha256:423d934c9fafb91aad38edf26efb46da91ffbc05f3f59c4b0c72e699720706f5", size = 40212062, upload-time = "2026-02-05T07:02:12.724Z" },
 ]
 
+[[package]]
+name = "openpyxl"
+version = "3.1.5"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "et-xmlfile" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/3d/f9/88d94a75de065ea32619465d2f77b29a0469500e99012523b91cc4141cd1/openpyxl-3.1.5.tar.gz", hash = "sha256:cf0e3cf56142039133628b5acffe8ef0c12bc902d2aadd3e0fe5878dc08d1050", size = 186464, upload-time = "2024-06-28T14:03:44.161Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c0/da/977ded879c29cbd04de313843e76868e6e13408a94ed6b987245dc7c8506/openpyxl-3.1.5-py2.py3-none-any.whl", hash = "sha256:5282c12b107bffeef825f4617dc029afaf41d0ea60823bbb665ef3079dc79de2", size = 250910, upload-time = "2024-06-28T14:03:41.161Z" },
+]
+
 [[package]]
 name = "orjson"
 version = "3.11.9"
@@ -2802,6 +2829,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/07/39/338d9219c4e87f3e708f18857ecd24d22a0c3094752393319553096b98af/scipy-1.17.1-cp314-cp314t-win_arm64.whl", hash = "sha256:200e1050faffacc162be6a486a984a0497866ec54149a01270adc8a59b7c7d21", size = 25489165, upload-time = "2026-02-23T00:22:29.563Z" },
 ]
 
+[[package]]
+name = "seaborn"
+version = "0.13.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "matplotlib" },
+    { name = "numpy" },
+    { name = "pandas" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/86/59/a451d7420a77ab0b98f7affa3a1d78a313d2f7281a57afb1a34bae8ab412/seaborn-0.13.2.tar.gz", hash = "sha256:93e60a40988f4d65e9f4885df477e2fdaff6b73a9ded434c1ab356dd57eefff7", size = 1457696, upload-time = "2024-01-25T13:21:52.551Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/83/11/00d3c3dfc25ad54e731d91449895a79e4bf2384dc3ac01809010ba88f6d5/seaborn-0.13.2-py3-none-any.whl", hash = "sha256:636f8336facf092165e27924f223d3c62ca560b1f2bb5dff7ab7fad265361987", size = 294914, upload-time = "2024-01-25T13:21:49.598Z" },
+]
+
 [[package]]
 name = "semantic-version"
 version = "2.10.0"