Refactor crop generation module as subpackage to avoid single module expanding indefinitely with added features

src/virtual_stain_flow/datasets/crop_dataset.py

@@ -2,7 +2,7 @@
 crop_dataset.py
 """
 
-from typing import Any, Dict, List, Sequence, Optional, Tuple, Union, Type
+from typing import Any, Dict, List, Sequence, Optional, Tuple, Union
 
 import numpy as np
 import pandas as pd
@@ -157,7 +157,7 @@ def from_base_dataset(
         cls,
         base_dataset: BaseImageDataset,
         transforms: Optional[Sequence[LoggableTransform]] = None,
-        how: Type[CropGenerator] = generate_center_crops,
+        how: CropGenerator = generate_center_crops,
         **kwargs: Any
     ) -> 'CropImageDataset':
         """

src/virtual_stain_flow/datasets/ds_engine/crop_generator.py

@@ -3,98 +3,22 @@
 
 Utilities for generating crop coordinates from BaseImageDataset objects.
 Designed for easy creation of CropImageDataset instances.
+Made Facade to account for increased complexity and future expansion.
 """
 
-from typing import Dict, List, Tuple, Any, Protocol
-
-from ..base_dataset import BaseImageDataset
-from .ds_utils import (
-    _get_active_channels, 
-    _validate_same_dimensions_across_channels
+from .crop_generators.protocol import CropSpec, CropMap, CropGenerator
+from .crop_generators.center import (
+    generate_center_crops,
+    _compute_center_crop,
 )
-
-CropSpec = Tuple[Tuple[int, int], int, int]
-CropMap = Dict[int, List[CropSpec]]
-
-
-class CropGenerator(Protocol):
-    """
-    Protocol for crop generator functions.
-    """
-    def __call__(
-        self, 
-        dataset: BaseImageDataset, 
-        **kwargs: Any
-    ) -> CropMap:
-        pass 
-
-
-def _compute_center_crop(
-    image_width: int,
-    image_height: int,
-    crop_size: int
-) -> Tuple[int, int]:
-    """
-    Compute top-left (x, y) coordinates for a center crop.
-
-    :param image_width: Width of the source image.
-    :param image_height: Height of the source image.
-    :param crop_size: Size of the square crop (width and height).
-    :return: Tuple of (x, y) for top-left corner of center crop.
-    :raises ValueError: If crop_size exceeds image dimensions.
-    """
-    if crop_size > image_width or crop_size > image_height:
-        raise ValueError(
-            f"crop_size ({crop_size}) exceeds image dimensions "
-            f"({image_width}x{image_height})."
-        )
-
-    x = (image_width - crop_size) // 2
-    y = (image_height - crop_size) // 2
-    return x, y
-
-
-def generate_center_crops(
-    dataset: BaseImageDataset,
-    crop_size: int,
-) -> CropMap:
-    """
-    Generate center crop coordinates for each sample in a BaseImageDataset.
-
-    :param dataset: A BaseImageDataset instance (or compatible object with
-        `file_state.manifest` attribute supporting `get_image_dimensions()`).
-    :param crop_size: Size of the square crop (same width and height).
-    :return: Dictionary mapping manifest indices to lists of crop specs.
-        Format: {manifest_idx: [((x, y), width, height), ...]}
-    :raises ValueError: If crop_size is non-positive, if no active channels
-        are configured, or if channel dimensions don't match for any sample.
-    """
-    if crop_size <= 0:
-        raise ValueError(f"crop_size must be positive, got {crop_size}.")
-
-    active_channels = _get_active_channels(dataset)
-    if not active_channels:
-        raise ValueError(
-            "No active channels configured. Set input_channel_keys and/or "
-            "target_channel_keys on the dataset before generating crops."
-        )
-
-    manifest = dataset.file_state.manifest
-    crop_specs: Dict[int, List[Tuple[Tuple[int, int], int, int]]] = {}
-
-    for idx in range(len(dataset)):
-        # Get dimensions for all active channels
-        dims = manifest.get_image_dimensions(idx, channels=active_channels)
-
-        # Validate all channels have matching dimensions
-        width, height = _validate_same_dimensions_across_channels(
-            dims, active_channels, idx
-        )
-
-        # Compute center crop coordinates
-        x, y = _compute_center_crop(width, height, crop_size)
-
-        # Store as crop_specs format: {idx: [((x, y), w, h), ...]}
-        crop_specs[idx] = [((x, y), crop_size, crop_size)]
-
-    return crop_specs
+from .crop_generators.point_centered import generate_point_centered_crops
+from .crop_generators.tile import generate_tile_crops
+__all__ = [
+    "CropSpec",
+    "CropMap",
+    "CropGenerator",
+    "generate_center_crops",
+    "generate_point_centered_crops",
+    "generate_tile_crops",
+    "_compute_center_crop",
+]

src/virtual_stain_flow/datasets/ds_engine/crop_generators/__init__.py

@@ -0,0 +1,14 @@
+from .protocol import CropMap, CropSpec, CropGenerator
+from .center import generate_center_crops, _compute_center_crop
+from .point_centered import generate_point_centered_crops
+from .tile import generate_tile_crops
+
+__all__ = [
+    "CropMap",
+    "CropSpec",
+    "CropGenerator",
+    "generate_center_crops",
+    "generate_point_centered_crops",
+    "generate_tile_crops",
+    "_compute_center_crop"
+]

src/virtual_stain_flow/datasets/ds_engine/crop_generators/center.py

@@ -0,0 +1,104 @@
+"""
+center.py
+
+Helper module for generating crops centered at FOV
+"""
+
+from typing import Dict, List, Tuple
+
+from .crop_summary import warn_formatted_crop_summary
+from .protocol import CropMap
+from ...base_dataset import BaseImageDataset
+from ..ds_utils import (
+    _get_active_channels, 
+    _validate_same_dimensions_across_channels
+)
+
+
+def _compute_center_crop(
+    image_width: int,
+    image_height: int,
+    crop_size: int
+) -> Tuple[int, int]:
+    """
+    Compute top-left (x, y) coordinates for a center crop.
+
+    :param image_width: Width of the source image.
+    :param image_height: Height of the source image.
+    :param crop_size: Size of the square crop (width and height).
+    :return: Tuple of (x, y) for top-left corner of center crop.
+    :raises ValueError: If crop_size exceeds image dimensions.
+    """
+    if crop_size > image_width or crop_size > image_height:
+        raise ValueError(
+            f"crop_size ({crop_size}) exceeds image dimensions "
+            f"({image_width}x{image_height})."
+        )
+
+    x = (image_width - crop_size) // 2
+    y = (image_height - crop_size) // 2
+    return x, y
+
+
+def generate_center_crops(
+    dataset: BaseImageDataset,
+    crop_size: int,
+    verbose: bool = False,
+) -> CropMap:
+    """
+    Generate center crop coordinates for each sample in a BaseImageDataset.
+
+    :param dataset: A BaseImageDataset instance (or compatible object with
+        `file_state.manifest` attribute supporting `get_image_dimensions()`).
+    :param crop_size: Size of the square crop (same width and height).
+    :param verbose: If True, emit summary statistics for crop generation.
+    :return: Dictionary mapping manifest indices to lists of crop specs.
+        Format: {manifest_idx: [((x, y), width, height), ...]}
+    :raises ValueError: If crop_size is non-positive, if no active channels
+        are configured, or if channel dimensions don't match for any sample.
+    """
+    if crop_size <= 0:
+        raise ValueError(f"crop_size must be positive, got {crop_size}.")
+
+    active_channels = _get_active_channels(dataset)
+    if not active_channels:
+        raise ValueError(
+            "No active channels configured. Set input_channel_keys and/or "
+            "target_channel_keys on the dataset before generating crops."
+        )
+
+    manifest = dataset.file_state.manifest
+    crop_specs: Dict[int, List[Tuple[Tuple[int, int], int, int]]] = {}
+    total_samples = len(dataset)
+
+    for idx in range(total_samples):
+        # Get dimensions for all active channels
+        dims = manifest.get_image_dimensions(idx, channels=active_channels)
+
+        # Validate all channels have matching dimensions
+        width, height = _validate_same_dimensions_across_channels(
+            dims, active_channels, idx
+        )
+
+        # Compute center crop coordinates
+        x, y = _compute_center_crop(width, height, crop_size)
+
+        # Store as crop_specs format: {idx: [((x, y), w, h), ...]}
+        crop_specs[idx] = [((x, y), crop_size, crop_size)]
+
+    if verbose:
+        successful_center_crops = len(crop_specs)
+        warn_formatted_crop_summary(
+            title="Center crop generation statistics:",
+            detail_line=(
+                "Generation criterion: exactly one centered crop is generated "
+                "per sample when crop size fits within image bounds."
+            ),
+            metrics={
+                "Success count / total dataset count": (
+                    f"{successful_center_crops}/{total_samples}"
+                ),
+            },
+        )
+
+    return crop_specs

src/virtual_stain_flow/datasets/ds_engine/crop_generators/crop_summary.py

@@ -0,0 +1,91 @@
+"""
+crop_summary.py
+
+Helper module for formatting crop generation summaries, 
+shared by crop generating modules
+"""
+
+from collections.abc import Mapping
+import warnings
+
+
+class CropSummaryWarning(UserWarning):
+	"""Warning category for crop generation summary reports."""
+
+
+def _stringify_metric_value(value: object) -> str:
+	"""Convert metric values to compact, human-readable strings."""
+	if isinstance(value, float):
+		return f"{value:.2f}"
+	return str(value)
+
+
+def build_stats_table(metrics: Mapping[str, object]) -> str:
+	"""
+	Build a simple ASCII table from metric-value key-value pairs.
+
+	:param metrics: Ordered metric mapping of label -> value.
+	:return: Multi-line ASCII table.
+	"""
+	metric_header = "Metric"
+	value_header = "Value"
+
+	rows = [(metric, _stringify_metric_value(value)) for metric, value in metrics.items()]
+	if not rows:
+		rows = [("No metrics", "n/a")]
+
+	metric_width = max(len(metric_header), *(len(metric) for metric, _ in rows))
+	value_width = max(len(value_header), *(len(value) for _, value in rows))
+
+	separator = f"+-{'-' * metric_width}-+-{'-' * value_width}-+"
+	header = f"| {metric_header:<{metric_width}} | {value_header:<{value_width}} |"
+	body = [
+		f"| {metric:<{metric_width}} | {value:>{value_width}} |"
+		for metric, value in rows
+	]
+
+	return "\n".join([separator, header, separator, *body, separator])
+
+
+def format_crop_summary(
+	title: str,
+	metrics: Mapping[str, object],
+	detail_line: str | None = None,
+) -> str:
+	"""
+	Format a crop summary title, optional detail line, and metric table.
+
+	:param title: Summary title line.
+	:param metrics: Ordered metric mapping of label -> value.
+	:param detail_line: Optional descriptive one-liner for context.
+	:return: Formatted summary text.
+	"""
+	stats_table = build_stats_table(metrics)
+	if detail_line:
+		return f"{title}\n{detail_line}\n{stats_table}"
+	return f"{title}\n{stats_table}"
+
+
+def warn_formatted_crop_summary(
+	title: str,
+	metrics: Mapping[str, object],
+	detail_line: str | None = None,
+) -> None:
+	"""
+	Emit a module-scoped crop summary warning without file/line prefixes.
+
+	:param title: Summary title line.
+	:param metrics: Ordered metric mapping of label -> value.
+	:param detail_line: Optional descriptive one-liner for context.
+	"""
+	message = format_crop_summary(title=title, metrics=metrics, detail_line=detail_line)
+
+	original_formatwarning = warnings.formatwarning
+	try:
+		warnings.formatwarning = (
+			lambda msg, category, filename, lineno, line=None:
+			f"{category.__name__}: {msg}\n"
+		)
+		warnings.warn(message, category=CropSummaryWarning, stacklevel=2)
+	finally:
+		warnings.formatwarning = original_formatwarning