AnomalyDetectionDataset

`anomaly_match.datasets.AnomalyDetectionDataset` ¶

Main dataset class for anomaly detection with labeled and unlabeled data.

`AnomalyDetectionDataset(cfg, data_source=None, labeled_cache_path=None)` ¶

Dataset for binary classification of normal vs anomaly images.

Handles labeled/unlabeled data management, train/test splitting, and label updates. Image loading is delegated to a TrainingDataSource.

Parameters:

Name	Type	Description	Default
`cfg`	`DotMap`	Configuration object containing dataset parameters.	required
`data_source`	`TrainingDataSource \| None`	Optional data source for loading images. If `None`, an `ImageFolderSource` is created from `cfg.data_dir`.	`None`
`labeled_cache_path`	`str \| None`	Optional path to a `LabeledDataCache` directory. When provided, labeled images are read from the cache and normalisation is applied at load time. The data_source is then used only for unlabeled sampling (should be created with `lazy=True` for container sources).	`None`

Source code in anomaly_match/datasets/AnomalyDetectionDataset.py

def __init__(
    self,
    cfg: DotMap,
    data_source: TrainingDataSource | None = None,
    labeled_cache_path: str | None = None,
) -> None:
    logger.debug(f"Loading AnomalyDetectionDataset from {cfg.data_dir}")

    self._labeled_cache_path = labeled_cache_path
    self.seed = cfg.seed
    self.size = cfg.normalisation.image_size
    self.root_dir = cfg.data_dir
    self.cfg = cfg
    self.test_ratio = cfg.test_ratio
    self.N_to_load = cfg.N_to_load
    self.label_file = (
        cfg.label_file if cfg.label_file else os.path.join(self.root_dir, "labeled_data.csv")
    )

    # Create data source (backward compatible default)
    if data_source is None:
        data_source = ImageFolderSource(cfg)
    self._data_source = data_source

    # For ImageFolderSource, expose discovered filenames; for other
    # sources the full filename list is built after loading.
    if isinstance(data_source, ImageFolderSource):
        self.all_filenames = data_source.all_filenames
    else:
        self.all_filenames = []

    # Auto-detect channel count from data source.  A channel_combination
    # matrix defines the output count by its rows (a 1x3 matrix turns RGB
    # into one channel), so raising n_output_channels to the source's band
    # count would contradict the images the matrix actually produces.
    detected_channels = data_source.detect_num_channels()
    if (
        detected_channels is not None
        and detected_channels > cfg.normalisation.n_output_channels
        and not isinstance(cfg.normalisation.channel_combination, (list, tuple, np.ndarray))
    ):
        logger.info(
            f"Detected {detected_channels} channels from images "
            f"(config had n_output_channels={cfg.normalisation.n_output_channels}), updating"
        )
        old_channels = cfg.normalisation.n_output_channels
        cfg.normalisation.n_output_channels = detected_channels
        cfg.num_channels = detected_channels
        for attr in ("norm_asinh_scale", "norm_asinh_clip"):
            val = cfg.normalisation[attr]
            if isinstance(val, list) and len(val) == old_channels:
                cfg.normalisation[attr] = val + [val[-1]] * (detected_channels - old_channels)
    self.num_channels = cfg.normalisation.n_output_channels

    self.N_to_load = min(self.N_to_load, data_source.get_total_count())

    # Initialize metadata handler
    self.metadata_handler = MetadataHandler(cfg.metadata_file, self.all_filenames)

    logger.info(f"Found {data_source.get_total_count()} total images in {self.root_dir}")

    self.split_indices = None
    self.data_dict: dict[str, tuple[np.ndarray, Label]] = {}

    # Load data via the data source
    self._load_from_source()

    # Load the labels from the CSV file
    self._load_csv_and_apply_labels()

    # Split the data into training and testing sets
    self._split_data()

`unlabeled_filepaths` `property` ¶

Return unlabeled filepaths.

`unlabeled_filenames` `property` ¶

Return unlabeled filenames.

`unlabeled` `property` ¶

Return unlabeled data in format [[imgs],[filenames]].

`train_data` `property` ¶

Return training data in format [[filenames],[imgs],[labels]].

`test_data` `property` ¶

Return testing data in format [[filenames],[imgs],[labels]].

`update_labels(new_labels_df)` ¶

Update the dataset with new labels for previously unlabeled images.

Parameters:

Name	Type	Description	Default
`new_labels_df`	`DataFrame`	DataFrame with an `id` (or `filename`) column and `label`.	required

Raises:

Type	Description
`ValueError`	If an invalid label is encountered.

Source code in anomaly_match/datasets/AnomalyDetectionDataset.py

def update_labels(self, new_labels_df: pd.DataFrame) -> None:
    """Update the dataset with new labels for previously unlabeled images.

    Args:
        new_labels_df: DataFrame with an ``id`` (or ``filename``) column and ``label``.

    Raises:
        ValueError: If an invalid label is encountered.
    """
    logger.debug(f"Updating labels for {new_labels_df.shape[0]} images")
    existing_values_changed = 0
    missing_filenames: list[str] = []

    for _, row in new_labels_df.iterrows():
        filename = os.path.basename(str(row["id"]))
        label = row["label"].lower()

        if label == LABEL_NORMAL:
            label_enum = Label.NORMAL
        elif label == LABEL_ANOMALY:
            label_enum = Label.ANOMALY
        elif label == LABEL_REMOVED:
            continue
        else:
            raise ValueError(f"Invalid label {label} for {filename}")

        if filename in self.data_dict:
            image, current_label = self.data_dict[filename]
            if current_label != label_enum and current_label != Label.UNKNOWN:
                self.data_dict[filename] = (image, label_enum)
                existing_values_changed += 1
            elif current_label == Label.UNKNOWN:
                self.data_dict[filename] = (image, label_enum)
        else:
            # Accumulate instead of logging per row — Cutana catalogues
            # with thousands of labels that didn't match the source
            # previously produced one WARNING line per orphan and
            # flooded the training screen.  One summary line at the
            # end is enough to flag the condition without the spam.
            missing_filenames.append(filename)

    logger.debug(f"Updated {existing_values_changed} existing labels")
    if missing_filenames:
        preview = ", ".join(missing_filenames[:3])
        if len(missing_filenames) > 3:
            preview += f" (+{len(missing_filenames) - 3} more)"
        logger.warning(
            "{} labelled id(s) not present in the loaded dataset — these rows "
            "are ignored for training: {}",
            len(missing_filenames),
            preview,
        )

`get_metadata_for_file(filename)` ¶

Get metadata for a specific file.

Parameters:

Name	Type	Description	Default
`filename`	`str`	The filename to get metadata for.	required

Returns:

Type	Description
`dict \| None`	Metadata for the file, or None if not found.

Source code in anomaly_match/datasets/AnomalyDetectionDataset.py

def get_metadata_for_file(self, filename: str) -> dict | None:
    """Get metadata for a specific file.

    Args:
        filename: The filename to get metadata for.

    Returns:
        Metadata for the file, or None if not found.
    """
    return self.metadata_handler.get_metadata_for_file(filename)

`get_all_metadata()` ¶

Get all metadata.

Returns:

Type	Description
`DataFrame \| None`	The full metadata DataFrame, or None if no metadata loaded.

Source code in anomaly_match/datasets/AnomalyDetectionDataset.py

def get_all_metadata(self) -> pd.DataFrame | None:
    """Get all metadata.

    Returns:
        The full metadata DataFrame, or None if no metadata loaded.
    """
    return self.metadata_handler.get_all_metadata()

AnomalyDetectionDataset

anomaly_match.datasets.AnomalyDetectionDataset ¶