Skip to content

training_data_source

anomaly_match.datasets.training_data_source

Abstract data source for training with multiple storage backends.

Provides a TrainingDataSource ABC that decouples image loading from the dataset labeling/splitting logic in AnomalyDetectionDataset. Concrete implementations handle image folders, Zarr arrays, and Cutana catalogues.

DataSourceType

Bases: str, Enum

Storage backend types for training data.

TrainingDataSource(cfg)

Bases: ABC

Abstract base class for loading training data from various storage formats.

Each implementation handles a different storage backend (filesystem images, Zarr arrays, Cutana catalogues). The source delivers preprocessed numpy arrays that AnomalyDetectionDataset can label and split.

Parameters:

Name Type Description Default
cfg DotMap

AnomalyMatch configuration with normalisation settings.

 required
Source code in anomaly_match/datasets/training_data_source.py 
54
55
56
57
58
59
60
61
def __init__(self, cfg: DotMap) -> None:
    self._cfg = cfg
    # Optional hook fired right after an unlabeled batch's source sizes are
    # recorded — before the slow cutout streaming.  The training subprocess
    # sets this to emit the size histogram early, so the plot is visible
    # while cutouts stream instead of only after.  Only Cutana records
    # sizes, so it's the only source that fires this.
    self.unlabeled_sizes_callback: Callable[[], None] | None = None

get_labeled_images(label_df) abstractmethod

Load images whose filenames appear in the label DataFrame.

Parameters:

Name Type Description Default
label_df DataFrame

DataFrame with at least filename and label columns.

 required

Returns:

Type Description
list[tuple[str, ndarray]]

List of (filename, image_array) tuples. image_array is HWC numpy,
list[tuple[str, ndarray]]

already resized and normalised per cfg.
Source code in anomaly_match/datasets/training_data_source.py 
63
64
65
66
67
68
69
70
71
72
73
@abstractmethod
def get_labeled_images(self, label_df: pd.DataFrame) -> list[tuple[str, np.ndarray]]:
    """Load images whose filenames appear in the label DataFrame.

    Args:
        label_df: DataFrame with at least ``filename`` and ``label`` columns.

    Returns:
        List of (filename, image_array) tuples. image_array is HWC numpy,
        already resized and normalised per cfg.
    """

get_unlabeled_batch(n, exclude) abstractmethod

Load a batch of up to n unlabeled images.

Parameters:

Name Type Description Default
n int

Maximum number of images to return.

 required
exclude set[str]

Filenames to skip (already labeled).

 required

Returns:

Type Description
list[tuple[str, ndarray]]

List of (filename, image_array) tuples.
Source code in anomaly_match/datasets/training_data_source.py 
75
76
77
78
79
80
81
82
83
84
85
@abstractmethod
def get_unlabeled_batch(self, n: int, exclude: set[str]) -> list[tuple[str, np.ndarray]]:
    """Load a batch of up to *n* unlabeled images.

    Args:
        n: Maximum number of images to return.
        exclude: Filenames to skip (already labeled).

    Returns:
        List of (filename, image_array) tuples.
    """

get_total_count() abstractmethod

Return the total number of images available in the data source.

Source code in anomaly_match/datasets/training_data_source.py 
87
88
89
@abstractmethod
def get_total_count(self) -> int:
    """Return the total number of images available in the data source."""

detect_num_channels() abstractmethod

Detect the number of image channels from a sample.

Returns:

Type Description
int | None

Number of channels, or None when detection is not applicable
int | None

(e.g. FITS files where channels depend on configuration).
Source code in anomaly_match/datasets/training_data_source.py 
91
92
93
94
95
96
97
98
@abstractmethod
def detect_num_channels(self) -> int | None:
    """Detect the number of image channels from a sample.

    Returns:
        Number of channels, or ``None`` when detection is not applicable
        (e.g. FITS files where channels depend on configuration).
    """

get_last_unlabeled_sizes()

Source sizes (diameters) of the most recent unlabeled batch.

Only Cutana catalogues carry a per-source size; other backends have no such column and return None. The training subprocess uses this to surface a size-distribution histogram of the sampled unlabeled pool.

Returns:

Type Description
ndarray | None

Finite source diameters from the last get_unlabeled_batch call,
ndarray | None

or None when the source exposes no size.
Source code in anomaly_match/datasets/training_data_source.py 
100
101
102
103
104
105
106
107
108
109
110
111
def get_last_unlabeled_sizes(self) -> np.ndarray | None:
    """Source sizes (diameters) of the most recent unlabeled batch.

    Only Cutana catalogues carry a per-source size; other backends have no
    such column and return ``None``.  The training subprocess uses this to
    surface a size-distribution histogram of the sampled unlabeled pool.

    Returns:
        Finite source diameters from the last ``get_unlabeled_batch`` call,
        or ``None`` when the source exposes no size.
    """
    return None

get_last_unlabeled_size_unit()

Unit of :meth:get_last_unlabeled_sizes ("pixel" / "arcsec").

Returns:

Type Description
str | None

The size unit, or None when the source exposes no size.
Source code in anomaly_match/datasets/training_data_source.py 
113
114
115
116
117
118
119
def get_last_unlabeled_size_unit(self) -> str | None:
    """Unit of :meth:`get_last_unlabeled_sizes` (``"pixel"`` / ``"arcsec"``).

    Returns:
        The size unit, or ``None`` when the source exposes no size.
    """
    return None

get_last_unlabeled_population_sizes()

Candidate-population diameters of the last unlabeled batch's tiles.

The population is every source in the tiles the sampler drew from — the baseline the training screen overlays the sampled pool against to show whether size stratification flattened the pool. Only Cutana exposes it; other backends return None.

Returns:

Type Description
ndarray | None

Finite population diameters, or None when the source is unsized.
Source code in anomaly_match/datasets/training_data_source.py 
121
122
123
124
125
126
127
128
129
130
131
132
def get_last_unlabeled_population_sizes(self) -> np.ndarray | None:
    """Candidate-population diameters of the last unlabeled batch's tiles.

    The population is every source in the tiles the sampler drew from — the
    baseline the training screen overlays the sampled pool against to show
    whether size stratification flattened the pool.  Only Cutana exposes it;
    other backends return ``None``.

    Returns:
        Finite population diameters, or ``None`` when the source is unsized.
    """
    return None

ImageFolderSource(cfg)

Bases: TrainingDataSource

Load training data from a directory of image files.

Parameters:

Name Type Description Default
cfg DotMap

Configuration; cfg.data_dir must point to an image directory.

 required
Source code in anomaly_match/datasets/training_data_source.py 
147
148
149
150
151
152
153
154
def __init__(self, cfg: DotMap) -> None:
    super().__init__(cfg)
    self._root_dir = cfg.data_dir
    self._all_filenames = get_image_names_from_folder(self._root_dir, recursive=False)

    rng = np.random.RandomState(cfg.seed)
    rng.shuffle(self._all_filenames)
    self._filename_set = set(self._all_filenames)

all_filenames property

All discovered image filenames (shuffled).

get_labeled_images(label_df)

Load all images whose filenames appear in label_df.

Returns:

Type Description
list[tuple[str, ndarray]]

List of (filename, image_array) tuples for matched files.
Source code in anomaly_match/datasets/training_data_source.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
def get_labeled_images(self, label_df: pd.DataFrame) -> list[tuple[str, np.ndarray]]:
    """Load all images whose filenames appear in *label_df*.

    Returns:
        List of (filename, image_array) tuples for matched files.
    """
    labeled_files = [f for f in label_df["id"] if f in self._filename_set]
    if not labeled_files:
        return []

    filepaths = [os.path.join(self._root_dir, f) for f in labeled_files]
    results = load_and_process_wrapper(filepaths, self._cfg, desc="Loading labeled images")
    return [(os.path.basename(fp), img) for fp, img in results]

get_unlabeled_batch(n, exclude)

Load up to n images that are not in exclude.

Returns:

Type Description
list[tuple[str, ndarray]]

List of (filename, image_array) tuples.
Source code in anomaly_match/datasets/training_data_source.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def get_unlabeled_batch(self, n: int, exclude: set[str]) -> list[tuple[str, np.ndarray]]:
    """Load up to *n* images that are not in *exclude*.

    Returns:
        List of (filename, image_array) tuples.
    """
    candidates = [f for f in self._all_filenames if f not in exclude]
    batch_files = candidates[:n]
    if not batch_files:
        return []

    filepaths = [os.path.join(self._root_dir, f) for f in batch_files]
    results = load_and_process_wrapper(filepaths, self._cfg, desc="Loading unlabeled batch")
    return [(os.path.basename(fp), img) for fp, img in results]

get_total_count()

Return number of image files in the directory.

Source code in anomaly_match/datasets/training_data_source.py 
190
191
192
def get_total_count(self) -> int:
    """Return number of image files in the directory."""
    return len(self._all_filenames)

detect_num_channels()

Auto-detect channels from a sample image.

Returns:

Type Description
int | None

Number of channels detected from the first image.
Source code in anomaly_match/datasets/training_data_source.py 
194
195
196
197
198
199
200
def detect_num_channels(self) -> int | None:
    """Auto-detect channels from a sample image.

    Returns:
        Number of channels detected from the first image.
    """
    return detect_num_channels(self._root_dir, self._all_filenames)

create_training_data_source(cfg, *, lazy=False)

Create the appropriate data source from configuration.

Uses cfg.training_data_source if set explicitly, otherwise auto-detects from cfg.data_dir.

Parameters:

Name Type Description Default
cfg DotMap

AnomalyMatch configuration.

 required
lazy bool

If True, skip O(N) indexing for container sources. Labeled images must come from a LabeledDataCache instead. Image folder sources ignore this flag.

 False

Returns:

Type Description
TrainingDataSource

A concrete TrainingDataSource instance.

Raises:

Type Description
ValueError

If the source type is unknown.
Source code in anomaly_match/datasets/training_data_source.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def create_training_data_source(cfg: DotMap, *, lazy: bool = False) -> TrainingDataSource:
    """Create the appropriate data source from configuration.

    Uses ``cfg.training_data_source`` if set explicitly, otherwise auto-detects
    from ``cfg.data_dir``.

    Args:
        cfg: AnomalyMatch configuration.
        lazy: If True, skip O(N) indexing for container sources. Labeled
            images must come from a ``LabeledDataCache`` instead. Image
            folder sources ignore this flag.

    Returns:
        A concrete ``TrainingDataSource`` instance.

    Raises:
        ValueError: If the source type is unknown.
    """
    source_type = cfg.training_data_source
    if source_type is None:
        source_type = _auto_detect_source_type(cfg.data_dir)

    logger.info("Training data source: {} (data_dir={})", source_type, cfg.data_dir)

    if source_type == DataSourceType.IMAGE_FOLDER:
        return ImageFolderSource(cfg)
    elif source_type == DataSourceType.ZARR:
        from anomaly_match.datasets.zarr_source import (  # noqa: PLC0415
            ZarrSource,  # lazy: avoid importing heavy optional dependency
        )

        return ZarrSource(cfg, lazy=lazy)
    elif source_type == DataSourceType.CUTANA:
        from anomaly_match.datasets.cutana_source import (  # noqa: PLC0415
            CutanaSource,  # lazy: avoid importing heavy optional dependency
        )

        return CutanaSource(cfg, lazy=lazy)
    else:
        raise ValueError(f"Unknown training data source type: {source_type!r}")