API Reference¶

Application Domain & Data¶

`app.domain.categories` ¶

Domain models and canonical category definitions for the MVTec AD dataset.

Provides centralized category constants, semantic splits (objects vs textures), and dynamic dataset discovery utilities.

`ANOMALY_DINO_MASKED_CATEGORIES: frozenset[str] = frozenset({'capsule', 'hazelnut', 'pill', 'screw', 'toothbrush'})` `module-attribute` ¶

Categories where background suppression/masking improves DINO representation.

`MVTEC_CATEGORIES: tuple[str, ...] = tuple(sorted(MVTEC_OBJECT_CATEGORIES + MVTEC_TEXTURE_CATEGORIES))` `module-attribute` ¶

All 15 official canonical benchmark categories of the MVTec AD dataset.

`MVTEC_OBJECT_CATEGORIES: tuple[str, ...] = ('bottle', 'cable', 'capsule', 'hazelnut', 'metal_nut', 'pill', 'screw', 'toothbrush', 'transistor', 'zipper')` `module-attribute` ¶

Rigid industrial object categories in MVTec AD requiring orientation preservation.

`MVTEC_TEXTURE_CATEGORIES: tuple[str, ...] = ('carpet', 'grid', 'leather', 'tile', 'wood')` `module-attribute` ¶

Spatially invariant surface texture categories in MVTec AD.

`OBJECT_CATEGORIES: frozenset[str] = frozenset(MVTEC_OBJECT_CATEGORIES)` `module-attribute` ¶

Set representation of rigid object categories for fast O(1) membership checks.

`TEXTURE_CATEGORIES: frozenset[str] = frozenset(MVTEC_TEXTURE_CATEGORIES)` `module-attribute` ¶

Set representation of surface texture categories for fast O(1) membership checks.

`discover_dataset_categories(data_root: str | Path | None = None) -> list[str]` ¶

Discover available MVTec categories dynamically from a dataset root directory.

If the specified directory exists and contains category subfolders, this function returns a sorted list of discovered folder names. If the directory does not exist, is unreadable, or contains no valid subfolders, it safely falls back to the canonical 15 MVTec AD benchmark categories.

Parameters:

Name	Type	Description	Default
`data_root`	`str \| Path \| None`	Path or string pointing to the dataset root folder.	`None`

Returns:

Type	Description
`list[str]`	Sorted list of discovered or fallback category names.

Source code in app/domain/categories.py

def discover_dataset_categories(data_root: str | Path | None = None) -> list[str]:
    """Discover available MVTec categories dynamically from a dataset root directory.

    If the specified directory exists and contains category subfolders, this function
    returns a sorted list of discovered folder names. If the directory does not exist,
    is unreadable, or contains no valid subfolders, it safely falls back to the
    canonical 15 MVTec AD benchmark categories.

    Args:
        data_root: Path or string pointing to the dataset root folder.

    Returns:
        Sorted list of discovered or fallback category names.
    """
    if data_root is None:
        return list(MVTEC_CATEGORIES)

    root_path = Path(data_root)
    if not root_path.is_dir():
        return list(MVTEC_CATEGORIES)

    discovered: list[str] = []
    try:
        for entry in root_path.iterdir():
            if entry.is_dir() and not entry.name.startswith((".", "_")):
                discovered.append(entry.name)
    except OSError:
        return list(MVTEC_CATEGORIES)

    if discovered:
        return sorted(discovered)

    return list(MVTEC_CATEGORIES)

`app.domain.data` ¶

Dataset helpers shared by modelling experiments.

`FairEvaluationSplit` `dataclass` ¶

Shared fitting, validation, and test partitions for one MVTec category.

Source code in app/domain/data.py

@dataclass(frozen=True)
class FairEvaluationSplit:
    """Shared fitting, validation, and test partitions for one MVTec category."""

    category: str
    fitting: pd.DataFrame
    validation: pd.DataFrame
    test: pd.DataFrame
    seed: int
    validation_fraction: float
    fitting_digest: str
    validation_digest: str
    test_digest: str

    @property
    def fitting_paths(self) -> list[str]:
        """Return fitting paths in their protocol-defined order."""
        return [str(path) for path in self.fitting["path"]]

    @property
    def validation_paths(self) -> list[str]:
        """Return validation paths in their protocol-defined order."""
        return [str(path) for path in self.validation["path"]]

    @property
    def test_paths(self) -> list[str]:
        """Return official test paths in their protocol-defined order."""
        return [str(path) for path in self.test["path"]]

    def evidence(self) -> dict[str, Any]:
        """Return serialisable protocol evidence for hashes and metadata."""
        return {
            "protocol": FAIR_EVALUATION_PROTOCOL,
            "split_seed": self.seed,
            "validation_fraction": self.validation_fraction,
            "train_normal": len(self.fitting),
            "val_normal": len(self.validation),
            "test_total": len(self.test),
            "fitting_path_digest": self.fitting_digest,
            "validation_path_digest": self.validation_digest,
            "test_path_digest": self.test_digest,
        }

`fitting_paths: list[str]` `property` ¶

Return fitting paths in their protocol-defined order.

`test_paths: list[str]` `property` ¶

Return official test paths in their protocol-defined order.

`validation_paths: list[str]` `property` ¶

Return validation paths in their protocol-defined order.

`evidence() -> dict[str, Any]` ¶

Return serialisable protocol evidence for hashes and metadata.

Source code in app/domain/data.py

def evidence(self) -> dict[str, Any]:
    """Return serialisable protocol evidence for hashes and metadata."""
    return {
        "protocol": FAIR_EVALUATION_PROTOCOL,
        "split_seed": self.seed,
        "validation_fraction": self.validation_fraction,
        "train_normal": len(self.fitting),
        "val_normal": len(self.validation),
        "test_total": len(self.test),
        "fitting_path_digest": self.fitting_digest,
        "validation_path_digest": self.validation_digest,
        "test_path_digest": self.test_digest,
    }

`MVTecImageDataset` ¶

Bases: Dataset[tuple[Tensor, int, str]]

Load manifest images as (tensor, anomaly label, path) tuples.

Attributes:

Name	Type	Description
`frame`		Manifest rows containing `path` and `is_anomaly`.
`transform`		Callable converting an RGB PIL image to a tensor.

Source code in app/domain/data.py

class MVTecImageDataset(Dataset[tuple[Tensor, int, str]]):
    """Load manifest images as ``(tensor, anomaly label, path)`` tuples.

    Attributes:
        frame: Manifest rows containing ``path`` and ``is_anomaly``.
        transform: Callable converting an RGB PIL image to a tensor.
    """

    def __init__(self, frame: pd.DataFrame, transform: Callable[[Image.Image], Tensor]) -> None:
        """Initialize the dataset from a manifest subset and image transform.

        Args:
            frame: Manifest rows containing ``path`` and ``is_anomaly``.
            transform: Callable converting an RGB PIL image to a tensor.

        Raises:
            ValueError: If a required manifest column is missing.
        """
        required_columns = {"path", "is_anomaly"}
        if missing_columns := required_columns.difference(frame.columns):
            logger.error("frame is missing required columns: %s", sorted(missing_columns))
            raise ValueError(f"frame is missing required columns: {sorted(missing_columns)}")

        self.frame = frame.loc[:, ["path", "is_anomaly"]].reset_index(drop=True).copy()
        self.transform = transform
        logger.info("Initialized dataset with %d images", len(self.frame))

    def __len__(self) -> int:
        """Return the number of manifest rows.

        Returns:
            Number of manifest rows.
        """
        return len(self.frame)

    def __getitem__(self, index: int) -> tuple[Tensor, int, str]:
        """Load and transform one image.

        Args:
            index: The index of the image to load.

        Returns:
            A tuple containing the transformed image tensor, the anomaly label, and the path to the image.
        """
        row = self.frame.iloc[index]
        path = str(row["path"])
        with Image.open(path) as image:
            image_tensor = self.transform(image.convert("RGB"))

        return image_tensor, int(row["is_anomaly"]), path

`getitem(index: int) -> tuple[Tensor, int, str]` ¶

Load and transform one image.

Parameters:

Name	Type	Description	Default
`index`	`int`	The index of the image to load.	required

Returns:

Type	Description
`tuple[Tensor, int, str]`	A tuple containing the transformed image tensor, the anomaly label, and the path to the image.

Source code in app/domain/data.py

def __getitem__(self, index: int) -> tuple[Tensor, int, str]:
    """Load and transform one image.

    Args:
        index: The index of the image to load.

    Returns:
        A tuple containing the transformed image tensor, the anomaly label, and the path to the image.
    """
    row = self.frame.iloc[index]
    path = str(row["path"])
    with Image.open(path) as image:
        image_tensor = self.transform(image.convert("RGB"))

    return image_tensor, int(row["is_anomaly"]), path

`init(frame: pd.DataFrame, transform: Callable[[Image.Image], Tensor]) -> None` ¶

Initialize the dataset from a manifest subset and image transform.

Parameters:

Name	Type	Description	Default
`frame`	`DataFrame`	Manifest rows containing `path` and `is_anomaly`.	required
`transform`	`Callable[[Image], Tensor]`	Callable converting an RGB PIL image to a tensor.	required

Raises:

Type	Description
`ValueError`	If a required manifest column is missing.

Source code in app/domain/data.py

def __init__(self, frame: pd.DataFrame, transform: Callable[[Image.Image], Tensor]) -> None:
    """Initialize the dataset from a manifest subset and image transform.

    Args:
        frame: Manifest rows containing ``path`` and ``is_anomaly``.
        transform: Callable converting an RGB PIL image to a tensor.

    Raises:
        ValueError: If a required manifest column is missing.
    """
    required_columns = {"path", "is_anomaly"}
    if missing_columns := required_columns.difference(frame.columns):
        logger.error("frame is missing required columns: %s", sorted(missing_columns))
        raise ValueError(f"frame is missing required columns: {sorted(missing_columns)}")

    self.frame = frame.loc[:, ["path", "is_anomaly"]].reset_index(drop=True).copy()
    self.transform = transform
    logger.info("Initialized dataset with %d images", len(self.frame))

`len() -> int` ¶

Return the number of manifest rows.

Returns:

Type	Description
`int`	Number of manifest rows.

Source code in app/domain/data.py

def __len__(self) -> int:
    """Return the number of manifest rows.

    Returns:
        Number of manifest rows.
    """
    return len(self.frame)

`build_fair_evaluation_split(manifest: pd.DataFrame, category: str, *, validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION, seed: int = FAIR_EVALUATION_SPLIT_SEED) -> FairEvaluationSplit` ¶

Build the deterministic shared baseline-evaluation split.

Only official normal training rows may enter fitting or validation. Official test rows retain the manifest's existing deterministic order.

Source code in app/domain/data.py

def build_fair_evaluation_split(
    manifest: pd.DataFrame,
    category: str,
    *,
    validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION,
    seed: int = FAIR_EVALUATION_SPLIT_SEED,
) -> FairEvaluationSplit:
    """Build the deterministic shared baseline-evaluation split.

    Only official normal training rows may enter fitting or validation. Official
    test rows retain the manifest's existing deterministic order.
    """
    from sklearn.model_selection import train_test_split

    required = {"path", "product", "split", "is_anomaly"}
    if missing := required.difference(manifest.columns):
        raise ValueError(f"manifest is missing required columns: {sorted(missing)}")
    if not 0.0 < validation_fraction < 1.0:
        raise ValueError("validation_fraction must be between 0 and 1")

    category_rows = manifest.loc[manifest["product"] == category].copy()
    if category_rows.empty:
        raise ValueError(f"No manifest rows found for category '{category}'")

    normal_train = category_rows.loc[
        (category_rows["split"] == "train") & (~category_rows["is_anomaly"].astype(bool))
    ].sort_values("path", kind="stable")
    official_test = category_rows.loc[category_rows["split"] == "test"]
    if normal_train.empty:
        raise ValueError(f"No official normal training rows found for category '{category}'")
    if official_test.empty:
        raise ValueError(f"No official test rows found for category '{category}'")

    fitting, validation = train_test_split(
        normal_train,
        test_size=validation_fraction,
        random_state=seed,
        shuffle=True,
    )
    fitting = fitting.reset_index(drop=True)
    validation = validation.reset_index(drop=True)
    official_test = official_test.reset_index(drop=True)

    fitting_paths = fitting["path"].astype(str).tolist()
    validation_paths = validation["path"].astype(str).tolist()
    test_paths = official_test["path"].astype(str).tolist()
    fitting_set = set(fitting_paths)
    validation_set = set(validation_paths)
    test_set = set(test_paths)
    if fitting_set & validation_set or fitting_set & test_set or validation_set & test_set:
        raise ValueError("Fair-evaluation partitions contain overlapping image paths")
    if fitting_set | validation_set != set(normal_train["path"].astype(str)):
        raise ValueError("Fitting and validation partitions do not exhaust normal training rows")

    return FairEvaluationSplit(
        category=category,
        fitting=fitting,
        validation=validation,
        test=official_test,
        seed=seed,
        validation_fraction=validation_fraction,
        fitting_digest=_ordered_path_digest(fitting_paths),
        validation_digest=_ordered_path_digest(validation_paths),
        test_digest=_ordered_path_digest(test_paths),
    )

`build_mvtec_manifest(root: str | Path) -> pd.DataFrame` ¶

Build a deterministic manifest of MVTec AD train and test images.

Ground-truth masks are linked through mask_path rather than included as samples. A missing mask is represented by None.

Parameters:

Name	Type	Description	Default
`root`	`str \| Path`	Directory containing MVTec product directories.	required

Returns:

Type	Description
`DataFrame`	One row per input image.

Raises:

Type	Description
`FileNotFoundError`	If the dataset root does not exist.
`NotADirectoryError`	If the dataset root is not a directory.
`ValueError`	If an image is unreadable or no images are found.

Source code in app/domain/data.py

def build_mvtec_manifest(root: str | Path) -> pd.DataFrame:
    """Build a deterministic manifest of MVTec AD train and test images.

    Ground-truth masks are linked through ``mask_path`` rather than included as
    samples. A missing mask is represented by ``None``.

    Args:
        root: Directory containing MVTec product directories.

    Returns:
        One row per input image.

    Raises:
        FileNotFoundError: If the dataset root does not exist.
        NotADirectoryError: If the dataset root is not a directory.
        ValueError: If an image is unreadable or no images are found.
    """
    logger.info("Building manifest for MVTec dataset at %s", root)
    dataset_root = Path(root).expanduser()
    if not dataset_root.exists():
        logger.error("MVTec dataset directory does not exist: %s", dataset_root)
        raise FileNotFoundError(f"MVTec dataset directory does not exist: {dataset_root}")
    if not dataset_root.is_dir():
        logger.error("MVTec dataset path is not a directory: %s", dataset_root)
        raise NotADirectoryError(f"MVTec dataset path is not a directory: {dataset_root}")
    dataset_root = dataset_root.resolve()

    rows: list[dict[str, object]] = []

    all_images = (
        path for path in dataset_root.rglob("*") if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS
    )

    for image_path in all_images:
        split = image_path.parent.parent.name
        if split not in ("train", "test"):
            continue

        product_dir = image_path.parent.parent.parent
        defect_dir = image_path.parent

        is_anomaly = defect_dir.name != "good"
        width, height, mode = _read_image_metadata(image_path)
        mask_path = product_dir / "ground_truth" / defect_dir.name / f"{image_path.stem}_mask.png"
        rows.append(
            {
                "path": str(image_path.resolve()),
                "image_id": image_path.stem,
                "product": product_dir.name,
                "split": split,
                "defect_type": defect_dir.name,
                "is_anomaly": is_anomaly,
                "width": width,
                "height": height,
                "mode": mode,
                "mask_path": str(mask_path.resolve()) if mask_path.is_file() else None,
            }
        )

    # Sort rows to adhere to your test assertion (train before test, etc.):
    rows.sort(
        key=lambda row: (
            row["product"],
            0 if row["split"] == "train" else 1,
            row["defect_type"],
            row["path"],
        )
    )

    if not rows:
        logger.error("No MVTec images were found below: %s", dataset_root)
        raise ValueError(f"No MVTec images were found below: {dataset_root}")

    logger.info("Built manifest: %d images across %d categories", len(rows), len({r["product"] for r in rows}))
    return pd.DataFrame(rows, columns=MANIFEST_COLUMNS)

`create_mvtec_dataset(manifest: pd.DataFrame, preprocessing_steps: list[dict[str, Any]] | None = None, image_size: tuple[int, int] = (256, 256)) -> MVTecImageDataset` ¶

Create an MVTecImageDataset from a manifest and preprocessing steps.

Parameters:

Name	Type	Description	Default
`manifest`	`DataFrame`	The manifest DataFrame.	required
`preprocessing_steps`	`list[dict[str, Any]] \| None`	A list of preprocessing steps.	`None`
`image_size`	`tuple[int, int]`	The size of the images.	`(256, 256)`

Returns:

Type	Description
`MVTecImageDataset`	An MVTecImageDataset.

Source code in app/domain/data.py

def create_mvtec_dataset(
    manifest: pd.DataFrame,
    preprocessing_steps: list[dict[str, Any]] | None = None,
    image_size: tuple[int, int] = (256, 256),
) -> MVTecImageDataset:
    """Create an MVTecImageDataset from a manifest and preprocessing steps.

    Args:
        manifest: The manifest DataFrame.
        preprocessing_steps: A list of preprocessing steps.
        image_size: The size of the images.

    Returns:
        An MVTecImageDataset.
    """
    from app.pipelines.preprocessing import PreprocessingTransformAdapter, build_pipeline_from_configs

    pipeline = build_pipeline_from_configs(preprocessing_steps)
    adapter = PreprocessingTransformAdapter(pipeline)

    transform = transforms.Compose(
        [
            transforms.Resize(image_size),
            adapter,
            transforms.ToTensor(),
        ]
    )

    return MVTecImageDataset(frame=manifest, transform=transform)

Command Line Interface¶

`app.cli` ¶

A command line interface for running pipelines.

This module provides a CLI for running different anomaly detection pipelines.

`main() -> None` ¶

Run the main CLI.

Source code in app/cli.py

def main() -> None:
    """Run the main CLI."""
    preprocess_sys_argv()

    parser = argparse.ArgumentParser(description="Industrial Component Anomaly Detection CLI")
    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    _setup_dummy_parser(subparsers)
    _setup_patchcore_parser(subparsers)
    _setup_cae_parser(subparsers)
    _setup_dinov2_parser(subparsers)
    _setup_dinov3_parser(subparsers)

    args = parser.parse_args()

    if args.command == "dummy":
        _handle_dummy_command(args)
    elif args.command in ("patchcore", "baseline"):
        _handle_patchcore_command(args)
    elif args.command == "cae":
        _handle_cae_command(args)
    elif args.command == "dinov2":
        _handle_dinov2_command(args)
    elif args.command == "dinov3":
        _handle_dinov3_command(args)
    else:
        parser.print_help()
        sys.exit(1)

`preprocess_sys_argv() -> None` ¶

Preprocess sys.argv to convert key=value positional args to --key value flags.

Source code in app/cli.py

def preprocess_sys_argv() -> None:
    """Preprocess sys.argv to convert key=value positional args to --key value flags."""
    new_argv = [sys.argv[0]]
    for arg in sys.argv[1:]:
        if "=" in arg and not arg.startswith("-"):
            key, val = arg.split("=", 1)
            flag = f"--{key.replace('_', '-')}"
            new_argv.extend([flag, val])
        else:
            new_argv.append(arg)
    sys.argv = new_argv

Modelling Pipelines¶

`app.pipelines.modelling.patchcore` ¶

Canonical PatchCore modelling pipeline subpackage.

`BaselineResult` ¶

Bases: TypedDict

Schema for overall PatchCore execution results.

Attributes:

Name	Type	Description
`category`	`str`	The specific category being evaluated.
`image_level`	`MetricLevelResult`	Image-level evaluation metrics.
`pixel_level`	`MetricLevelResult`	Pixel-level evaluation metrics.
`raw_results`	`dict[str, float]`	Raw evaluation results from the Anomalib engine.
`heatmap_overlays`	`dict[int, dict[str, list[Any]]]`	Dictionary of generated heatmap overlays.
`anomalous_indices`	`list[int]`	List of test dataset indices that are anomalous.
`preprocessing_steps`	`list[dict[str, Any]]`	List of active preprocessing step configurations.
`hyperparameters`	`dict[str, Any]`	Dictionary of model hyperparameters.
`dataset_split`	`dict[str, Any]`	Dictionary of dataset partition sample counts.
`model_hash`	`str`	Unique 12-char model hash.
`metadata`	`dict[str, Any]`	Full metadata dictionary.

Source code in app/domain/evaluation.py

class BaselineResult(TypedDict, total=False):
    """Schema for overall PatchCore execution results.

    Attributes:
        category: The specific category being evaluated.
        image_level: Image-level evaluation metrics.
        pixel_level: Pixel-level evaluation metrics.
        raw_results: Raw evaluation results from the Anomalib engine.
        heatmap_overlays: Dictionary of generated heatmap overlays.
        anomalous_indices: List of test dataset indices that are anomalous.
        preprocessing_steps: List of active preprocessing step configurations.
        hyperparameters: Dictionary of model hyperparameters.
        dataset_split: Dictionary of dataset partition sample counts.
        model_hash: Unique 12-char model hash.
        metadata: Full metadata dictionary.
    """

    category: str
    image_level: MetricLevelResult
    pixel_level: MetricLevelResult
    raw_results: dict[str, float]
    heatmap_overlays: dict[int, dict[str, list[Any]]]
    anomalous_indices: list[int]
    preprocessing_steps: list[dict[str, Any]]
    hyperparameters: dict[str, Any]
    dataset_split: dict[str, Any]
    model_hash: str
    metadata: dict[str, Any]

`ConfusionMatrix` `dataclass` ¶

Confusion counts for anomalous detection at calibrated threshold.

Attributes:

Name	Type	Description
`true_positives`	`int`	Count of correctly identified defective components.
`false_positives`	`int`	Count of healthy components incorrectly flagged.
`false_negatives`	`int`	Count of defective components missed.
`true_negatives`	`int`	Count of healthy components correctly identified.

Source code in app/domain/evaluation.py

@dataclass(frozen=True)
class ConfusionMatrix:
    """Confusion counts for anomalous detection at calibrated threshold.

    Attributes:
        true_positives: Count of correctly identified defective components.
        false_positives: Count of healthy components incorrectly flagged.
        false_negatives: Count of defective components missed.
        true_negatives: Count of healthy components correctly identified.
    """

    true_positives: int
    false_positives: int
    false_negatives: int
    true_negatives: int

`EvaluationArtifacts` `dataclass` ¶

Structured evaluation artifacts, metrics, and visual overlays.

Attributes:

Name	Type	Description
`image_metrics`	`ImageEvaluationMetrics`	Detailed image-level classification performance metrics.
`pixel_metrics`	`PixelEvaluationMetrics`	Detailed pixel-level localization performance metrics.
`thresholds`	`Thresholds`	Calibrated frozen thresholds.
`heatmap_overlays`	`dict[int, dict[str, list[Any]]]`	Dictionary mapping test indices to overlay visual arrays.
`anomalous_indices`	`list[int]`	List of test dataset indices that are ground-truth anomalous.

Source code in app/domain/evaluation.py

@dataclass(frozen=True)
class EvaluationArtifacts:
    """Structured evaluation artifacts, metrics, and visual overlays.

    Attributes:
        image_metrics: Detailed image-level classification performance metrics.
        pixel_metrics: Detailed pixel-level localization performance metrics.
        thresholds: Calibrated frozen thresholds.
        heatmap_overlays: Dictionary mapping test indices to overlay visual arrays.
        anomalous_indices: List of test dataset indices that are ground-truth anomalous.
    """

    image_metrics: ImageEvaluationMetrics
    pixel_metrics: PixelEvaluationMetrics
    thresholds: Thresholds
    heatmap_overlays: dict[int, dict[str, list[Any]]] = field(default_factory=dict)
    anomalous_indices: list[int] = field(default_factory=list)

    def __len__(self) -> int:
        """Return length of legacy tuple representation."""
        return 18

    def __getitem__(self, index: int | slice) -> Any:
        """Support indexing and slicing for legacy tuple compatibility."""
        return tuple(self)[index]

    def __iter__(self) -> Iterator[Any]:
        """Support backwards-compatible unpacking into the legacy 18-tuple."""
        return iter(
            (
                self.image_metrics.f1_score,
                self.pixel_metrics.f1_score,
                self.image_metrics.precision,
                self.image_metrics.recall,
                self.thresholds.image,
                self.pixel_metrics.auroc,
                self.pixel_metrics.aupimo,
                self.pixel_metrics.anomaly_map_min,
                self.pixel_metrics.anomaly_map_max,
                self.pixel_metrics.anomaly_map_range,
                self.heatmap_overlays,
                self.anomalous_indices,
                self.image_metrics.confusion.true_positives,
                self.image_metrics.confusion.false_positives,
                self.image_metrics.confusion.false_negatives,
                self.image_metrics.confusion.true_negatives,
                self.thresholds.pixel,
                self.image_metrics.auroc,
            )
        )

    @classmethod
    def from_tuple(cls, values: Any) -> "EvaluationArtifacts":
        """Construct EvaluationArtifacts from a legacy 18-element tuple or return if already EvaluationArtifacts.

        Args:
            values: An EvaluationArtifacts instance or a sequence of 18 values matching the legacy tuple layout.

        Returns:
            An EvaluationArtifacts instance.

        Raises:
            ValueError: If values is not an EvaluationArtifacts and does not have exactly 18 elements.
        """
        if isinstance(values, cls):
            return values
        raw = tuple(values)
        if len(raw) != 18:
            raise ValueError(f"Expected 18 elements for legacy evaluation tuple, got {len(raw)}")
        (
            image_f1,
            pixel_f1,
            image_precision,
            image_recall,
            image_threshold,
            pixel_auroc,
            pixel_aupimo,
            anomaly_map_min,
            anomaly_map_max,
            anomaly_map_range,
            heatmap_overlays,
            anomalous_indices,
            true_positives,
            false_positives,
            false_negatives,
            true_negatives,
            pixel_threshold,
            image_auroc,
        ) = raw

        confusion = ConfusionMatrix(
            true_positives=int(true_positives),
            false_positives=int(false_positives),
            false_negatives=int(false_negatives),
            true_negatives=int(true_negatives),
        )
        thresholds = Thresholds(
            image=float(image_threshold),
            pixel=float(pixel_threshold),
        )
        image_metrics = ImageEvaluationMetrics(
            auroc=float(image_auroc),
            f1_score=float(image_f1),
            precision=float(image_precision),
            recall=float(image_recall),
            threshold=float(image_threshold),
            confusion=confusion,
        )
        pixel_metrics = PixelEvaluationMetrics(
            auroc=float(pixel_auroc),
            aupimo=float(pixel_aupimo),
            f1_score=float(pixel_f1),
            threshold=float(pixel_threshold),
            anomaly_map_min=float(anomaly_map_min),
            anomaly_map_max=float(anomaly_map_max),
            anomaly_map_range=float(anomaly_map_range),
        )
        return cls(
            image_metrics=image_metrics,
            pixel_metrics=pixel_metrics,
            thresholds=thresholds,
            heatmap_overlays=heatmap_overlays,
            anomalous_indices=anomalous_indices,
        )

`getitem(index: int | slice) -> Any` ¶

Support indexing and slicing for legacy tuple compatibility.

Source code in app/domain/evaluation.py

def __getitem__(self, index: int | slice) -> Any:
    """Support indexing and slicing for legacy tuple compatibility."""
    return tuple(self)[index]

`iter() -> Iterator[Any]` ¶

Support backwards-compatible unpacking into the legacy 18-tuple.

Source code in app/domain/evaluation.py

def __iter__(self) -> Iterator[Any]:
    """Support backwards-compatible unpacking into the legacy 18-tuple."""
    return iter(
        (
            self.image_metrics.f1_score,
            self.pixel_metrics.f1_score,
            self.image_metrics.precision,
            self.image_metrics.recall,
            self.thresholds.image,
            self.pixel_metrics.auroc,
            self.pixel_metrics.aupimo,
            self.pixel_metrics.anomaly_map_min,
            self.pixel_metrics.anomaly_map_max,
            self.pixel_metrics.anomaly_map_range,
            self.heatmap_overlays,
            self.anomalous_indices,
            self.image_metrics.confusion.true_positives,
            self.image_metrics.confusion.false_positives,
            self.image_metrics.confusion.false_negatives,
            self.image_metrics.confusion.true_negatives,
            self.thresholds.pixel,
            self.image_metrics.auroc,
        )
    )

`len() -> int` ¶

Return length of legacy tuple representation.

Source code in app/domain/evaluation.py

def __len__(self) -> int:
    """Return length of legacy tuple representation."""
    return 18

`from_tuple(values: Any) -> EvaluationArtifacts` `classmethod` ¶

Construct EvaluationArtifacts from a legacy 18-element tuple or return if already EvaluationArtifacts.

Parameters:

Name	Type	Description	Default
`values`	`Any`	An EvaluationArtifacts instance or a sequence of 18 values matching the legacy tuple layout.	required

Returns:

Type	Description
`EvaluationArtifacts`	An EvaluationArtifacts instance.

Raises:

Type	Description
`ValueError`	If values is not an EvaluationArtifacts and does not have exactly 18 elements.

Source code in app/domain/evaluation.py

@classmethod
def from_tuple(cls, values: Any) -> "EvaluationArtifacts":
    """Construct EvaluationArtifacts from a legacy 18-element tuple or return if already EvaluationArtifacts.

    Args:
        values: An EvaluationArtifacts instance or a sequence of 18 values matching the legacy tuple layout.

    Returns:
        An EvaluationArtifacts instance.

    Raises:
        ValueError: If values is not an EvaluationArtifacts and does not have exactly 18 elements.
    """
    if isinstance(values, cls):
        return values
    raw = tuple(values)
    if len(raw) != 18:
        raise ValueError(f"Expected 18 elements for legacy evaluation tuple, got {len(raw)}")
    (
        image_f1,
        pixel_f1,
        image_precision,
        image_recall,
        image_threshold,
        pixel_auroc,
        pixel_aupimo,
        anomaly_map_min,
        anomaly_map_max,
        anomaly_map_range,
        heatmap_overlays,
        anomalous_indices,
        true_positives,
        false_positives,
        false_negatives,
        true_negatives,
        pixel_threshold,
        image_auroc,
    ) = raw

    confusion = ConfusionMatrix(
        true_positives=int(true_positives),
        false_positives=int(false_positives),
        false_negatives=int(false_negatives),
        true_negatives=int(true_negatives),
    )
    thresholds = Thresholds(
        image=float(image_threshold),
        pixel=float(pixel_threshold),
    )
    image_metrics = ImageEvaluationMetrics(
        auroc=float(image_auroc),
        f1_score=float(image_f1),
        precision=float(image_precision),
        recall=float(image_recall),
        threshold=float(image_threshold),
        confusion=confusion,
    )
    pixel_metrics = PixelEvaluationMetrics(
        auroc=float(pixel_auroc),
        aupimo=float(pixel_aupimo),
        f1_score=float(pixel_f1),
        threshold=float(pixel_threshold),
        anomaly_map_min=float(anomaly_map_min),
        anomaly_map_max=float(anomaly_map_max),
        anomaly_map_range=float(anomaly_map_range),
    )
    return cls(
        image_metrics=image_metrics,
        pixel_metrics=pixel_metrics,
        thresholds=thresholds,
        heatmap_overlays=heatmap_overlays,
        anomalous_indices=anomalous_indices,
    )

`FairEvaluationSplit` `dataclass` ¶

Shared fitting, validation, and test partitions for one MVTec category.

Source code in app/domain/data.py

@dataclass(frozen=True)
class FairEvaluationSplit:
    """Shared fitting, validation, and test partitions for one MVTec category."""

    category: str
    fitting: pd.DataFrame
    validation: pd.DataFrame
    test: pd.DataFrame
    seed: int
    validation_fraction: float
    fitting_digest: str
    validation_digest: str
    test_digest: str

    @property
    def fitting_paths(self) -> list[str]:
        """Return fitting paths in their protocol-defined order."""
        return [str(path) for path in self.fitting["path"]]

    @property
    def validation_paths(self) -> list[str]:
        """Return validation paths in their protocol-defined order."""
        return [str(path) for path in self.validation["path"]]

    @property
    def test_paths(self) -> list[str]:
        """Return official test paths in their protocol-defined order."""
        return [str(path) for path in self.test["path"]]

    def evidence(self) -> dict[str, Any]:
        """Return serialisable protocol evidence for hashes and metadata."""
        return {
            "protocol": FAIR_EVALUATION_PROTOCOL,
            "split_seed": self.seed,
            "validation_fraction": self.validation_fraction,
            "train_normal": len(self.fitting),
            "val_normal": len(self.validation),
            "test_total": len(self.test),
            "fitting_path_digest": self.fitting_digest,
            "validation_path_digest": self.validation_digest,
            "test_path_digest": self.test_digest,
        }

`fitting_paths: list[str]` `property` ¶

Return fitting paths in their protocol-defined order.

`test_paths: list[str]` `property` ¶

Return official test paths in their protocol-defined order.

`validation_paths: list[str]` `property` ¶

Return validation paths in their protocol-defined order.

`evidence() -> dict[str, Any]` ¶

Return serialisable protocol evidence for hashes and metadata.

Source code in app/domain/data.py

def evidence(self) -> dict[str, Any]:
    """Return serialisable protocol evidence for hashes and metadata."""
    return {
        "protocol": FAIR_EVALUATION_PROTOCOL,
        "split_seed": self.seed,
        "validation_fraction": self.validation_fraction,
        "train_normal": len(self.fitting),
        "val_normal": len(self.validation),
        "test_total": len(self.test),
        "fitting_path_digest": self.fitting_digest,
        "validation_path_digest": self.validation_digest,
        "test_path_digest": self.test_digest,
    }

`ImageEvaluationMetrics` `dataclass` ¶

Image-level anomaly classification metrics.

Attributes:

Name	Type	Description
`auroc`	`float`	Area under the Receiver Operating Characteristic curve.
`f1_score`	`float`	Harmonic mean of precision and recall at calibrated threshold.
`precision`	`float`	Fraction of flagged components that are truly defective.
`recall`	`float`	Fraction of defective components correctly detected.
`threshold`	`float`	Frozen classification threshold applied.
`confusion`	`ConfusionMatrix`	Full confusion matrix counts.

Source code in app/domain/evaluation.py

@dataclass(frozen=True)
class ImageEvaluationMetrics:
    """Image-level anomaly classification metrics.

    Attributes:
        auroc: Area under the Receiver Operating Characteristic curve.
        f1_score: Harmonic mean of precision and recall at calibrated threshold.
        precision: Fraction of flagged components that are truly defective.
        recall: Fraction of defective components correctly detected.
        threshold: Frozen classification threshold applied.
        confusion: Full confusion matrix counts.
    """

    auroc: float
    f1_score: float
    precision: float
    recall: float
    threshold: float
    confusion: ConfusionMatrix

`MetricLevelResult` ¶

Bases: TypedDict

Schema for individual evaluation level metrics (image or pixel).

Attributes:

Name	Type	Description
`auroc`	`float`	Area Under the Receiver Operating Characteristic Curve.
`average_precision`	`float`	Average Precision score.
`f1_score`	`float`	F1 score for the given metric level.
`precision`	`float`	Precision score.
`recall`	`float`	Recall score.
`threshold`	`float`	Decision threshold for classification.
`aupimo_score`	`float`	Integrated AUPIMO score.
`fpr_lower_bound`	`float`	Lower bound for FPR integration.
`fpr_upper_bound`	`float`	Upper bound for FPR integration.
`aupimo`	`float`	AUPIMO score.
`anomaly_map_min`	`float`	Minimum anomaly-map score over the test set.
`anomaly_map_max`	`float`	Maximum anomaly-map score over the test set.
`anomaly_map_range`	`float`	Difference between maximum and minimum map scores.
`metrics_path`	`str`	Path to the .npz file containing precision, recall, and thresholds.
`true_positives`	`int`	Count of true positive predictions.
`false_positives`	`int`	Count of false positive predictions.
`false_negatives`	`int`	Count of false negative predictions.
`true_negatives`	`int`	Count of true negative predictions.
`aupimo_num_thresholds`	`int`	Number of thresholds used in AUPIMO integration.
`canonical_height`	`int`	Canonical evaluation map height.
`canonical_width`	`int`	Canonical evaluation map width.

Source code in app/domain/evaluation.py

class MetricLevelResult(TypedDict, total=False):
    """Schema for individual evaluation level metrics (image or pixel).

    Attributes:
        auroc: Area Under the Receiver Operating Characteristic Curve.
        average_precision: Average Precision score.
        f1_score: F1 score for the given metric level.
        precision: Precision score.
        recall: Recall score.
        threshold: Decision threshold for classification.
        aupimo_score: Integrated AUPIMO score.
        fpr_lower_bound: Lower bound for FPR integration.
        fpr_upper_bound: Upper bound for FPR integration.
        aupimo: AUPIMO score.
        anomaly_map_min: Minimum anomaly-map score over the test set.
        anomaly_map_max: Maximum anomaly-map score over the test set.
        anomaly_map_range: Difference between maximum and minimum map scores.
        metrics_path: Path to the .npz file containing precision, recall, and thresholds.
        true_positives: Count of true positive predictions.
        false_positives: Count of false positive predictions.
        false_negatives: Count of false negative predictions.
        true_negatives: Count of true negative predictions.
        aupimo_num_thresholds: Number of thresholds used in AUPIMO integration.
        canonical_height: Canonical evaluation map height.
        canonical_width: Canonical evaluation map width.
    """

    auroc: float
    average_precision: float
    f1_score: float
    precision: float
    recall: float
    threshold: float
    aupimo_score: float
    fpr_lower_bound: float
    fpr_upper_bound: float
    aupimo: float
    anomaly_map_min: float
    anomaly_map_max: float
    anomaly_map_range: float
    metrics_path: str
    true_positives: int
    false_positives: int
    false_negatives: int
    true_negatives: int
    aupimo_num_thresholds: int
    canonical_height: int
    canonical_width: int

`PixelEvaluationMetrics` `dataclass` ¶

Pixel-level anomaly localization metrics computed on canonical 256x256 grids.

Attributes:

Name	Type	Description
`auroc`	`float`	Pixel-level Area under the ROC curve.
`aupimo`	`float`	Strict Area under the Per-Image Overlap curve within standard bounds.
`f1_score`	`float`	Pixel-level binary segmentation F1 score.
`threshold`	`float`	Frozen segmentation threshold applied.
`anomaly_map_min`	`float`	Minimum raw continuous score over test set.
`anomaly_map_max`	`float`	Maximum raw continuous score over test set.
`anomaly_map_range`	`float`	Continuous score spread (max - min).

Source code in app/domain/evaluation.py

@dataclass(frozen=True)
class PixelEvaluationMetrics:
    """Pixel-level anomaly localization metrics computed on canonical 256x256 grids.

    Attributes:
        auroc: Pixel-level Area under the ROC curve.
        aupimo: Strict Area under the Per-Image Overlap curve within standard bounds.
        f1_score: Pixel-level binary segmentation F1 score.
        threshold: Frozen segmentation threshold applied.
        anomaly_map_min: Minimum raw continuous score over test set.
        anomaly_map_max: Maximum raw continuous score over test set.
        anomaly_map_range: Continuous score spread (max - min).
    """

    auroc: float
    aupimo: float
    f1_score: float
    threshold: float
    anomaly_map_min: float
    anomaly_map_max: float
    anomaly_map_range: float

`Thresholds` `dataclass` ¶

Calibrated decision thresholds calibrated strictly on normal validation partition.

Attributes:

Name	Type	Description
`image`	`float`	Image-level continuous score classification threshold.
`pixel`	`float`	Pixel-level anomaly map localization threshold.

Source code in app/domain/evaluation.py

@dataclass(frozen=True)
class Thresholds:
    """Calibrated decision thresholds calibrated strictly on normal validation partition.

    Attributes:
        image: Image-level continuous score classification threshold.
        pixel: Pixel-level anomaly map localization threshold.
    """

    image: float
    pixel: float

`build_fair_evaluation_split(manifest: pd.DataFrame, category: str, *, validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION, seed: int = FAIR_EVALUATION_SPLIT_SEED) -> FairEvaluationSplit` ¶

Build the deterministic shared baseline-evaluation split.

Only official normal training rows may enter fitting or validation. Official test rows retain the manifest's existing deterministic order.

Source code in app/domain/data.py

def build_fair_evaluation_split(
    manifest: pd.DataFrame,
    category: str,
    *,
    validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION,
    seed: int = FAIR_EVALUATION_SPLIT_SEED,
) -> FairEvaluationSplit:
    """Build the deterministic shared baseline-evaluation split.

    Only official normal training rows may enter fitting or validation. Official
    test rows retain the manifest's existing deterministic order.
    """
    from sklearn.model_selection import train_test_split

    required = {"path", "product", "split", "is_anomaly"}
    if missing := required.difference(manifest.columns):
        raise ValueError(f"manifest is missing required columns: {sorted(missing)}")
    if not 0.0 < validation_fraction < 1.0:
        raise ValueError("validation_fraction must be between 0 and 1")

    category_rows = manifest.loc[manifest["product"] == category].copy()
    if category_rows.empty:
        raise ValueError(f"No manifest rows found for category '{category}'")

    normal_train = category_rows.loc[
        (category_rows["split"] == "train") & (~category_rows["is_anomaly"].astype(bool))
    ].sort_values("path", kind="stable")
    official_test = category_rows.loc[category_rows["split"] == "test"]
    if normal_train.empty:
        raise ValueError(f"No official normal training rows found for category '{category}'")
    if official_test.empty:
        raise ValueError(f"No official test rows found for category '{category}'")

    fitting, validation = train_test_split(
        normal_train,
        test_size=validation_fraction,
        random_state=seed,
        shuffle=True,
    )
    fitting = fitting.reset_index(drop=True)
    validation = validation.reset_index(drop=True)
    official_test = official_test.reset_index(drop=True)

    fitting_paths = fitting["path"].astype(str).tolist()
    validation_paths = validation["path"].astype(str).tolist()
    test_paths = official_test["path"].astype(str).tolist()
    fitting_set = set(fitting_paths)
    validation_set = set(validation_paths)
    test_set = set(test_paths)
    if fitting_set & validation_set or fitting_set & test_set or validation_set & test_set:
        raise ValueError("Fair-evaluation partitions contain overlapping image paths")
    if fitting_set | validation_set != set(normal_train["path"].astype(str)):
        raise ValueError("Fitting and validation partitions do not exhaust normal training rows")

    return FairEvaluationSplit(
        category=category,
        fitting=fitting,
        validation=validation,
        test=official_test,
        seed=seed,
        validation_fraction=validation_fraction,
        fitting_digest=_ordered_path_digest(fitting_paths),
        validation_digest=_ordered_path_digest(validation_paths),
        test_digest=_ordered_path_digest(test_paths),
    )

`build_mvtec_manifest(root: str | Path) -> pd.DataFrame` ¶

Build a deterministic manifest of MVTec AD train and test images.

Ground-truth masks are linked through mask_path rather than included as samples. A missing mask is represented by None.

Parameters:

Name	Type	Description	Default
`root`	`str \| Path`	Directory containing MVTec product directories.	required

Returns:

Type	Description
`DataFrame`	One row per input image.

Raises:

Type	Description
`FileNotFoundError`	If the dataset root does not exist.
`NotADirectoryError`	If the dataset root is not a directory.
`ValueError`	If an image is unreadable or no images are found.

Source code in app/domain/data.py

def build_mvtec_manifest(root: str | Path) -> pd.DataFrame:
    """Build a deterministic manifest of MVTec AD train and test images.

    Ground-truth masks are linked through ``mask_path`` rather than included as
    samples. A missing mask is represented by ``None``.

    Args:
        root: Directory containing MVTec product directories.

    Returns:
        One row per input image.

    Raises:
        FileNotFoundError: If the dataset root does not exist.
        NotADirectoryError: If the dataset root is not a directory.
        ValueError: If an image is unreadable or no images are found.
    """
    logger.info("Building manifest for MVTec dataset at %s", root)
    dataset_root = Path(root).expanduser()
    if not dataset_root.exists():
        logger.error("MVTec dataset directory does not exist: %s", dataset_root)
        raise FileNotFoundError(f"MVTec dataset directory does not exist: {dataset_root}")
    if not dataset_root.is_dir():
        logger.error("MVTec dataset path is not a directory: %s", dataset_root)
        raise NotADirectoryError(f"MVTec dataset path is not a directory: {dataset_root}")
    dataset_root = dataset_root.resolve()

    rows: list[dict[str, object]] = []

    all_images = (
        path for path in dataset_root.rglob("*") if path.is_file() and path.suffix.lower() in IMAGE_EXTENSIONS
    )

    for image_path in all_images:
        split = image_path.parent.parent.name
        if split not in ("train", "test"):
            continue

        product_dir = image_path.parent.parent.parent
        defect_dir = image_path.parent

        is_anomaly = defect_dir.name != "good"
        width, height, mode = _read_image_metadata(image_path)
        mask_path = product_dir / "ground_truth" / defect_dir.name / f"{image_path.stem}_mask.png"
        rows.append(
            {
                "path": str(image_path.resolve()),
                "image_id": image_path.stem,
                "product": product_dir.name,
                "split": split,
                "defect_type": defect_dir.name,
                "is_anomaly": is_anomaly,
                "width": width,
                "height": height,
                "mode": mode,
                "mask_path": str(mask_path.resolve()) if mask_path.is_file() else None,
            }
        )

    # Sort rows to adhere to your test assertion (train before test, etc.):
    rows.sort(
        key=lambda row: (
            row["product"],
            0 if row["split"] == "train" else 1,
            row["defect_type"],
            row["path"],
        )
    )

    if not rows:
        logger.error("No MVTec images were found below: %s", dataset_root)
        raise ValueError(f"No MVTec images were found below: {dataset_root}")

    logger.info("Built manifest: %d images across %d categories", len(rows), len({r["product"] for r in rows}))
    return pd.DataFrame(rows, columns=MANIFEST_COLUMNS)

`delete_cached_patchcore_model(model_hash: str, registry_base: str | Path, soft_delete: bool = True) -> bool` ¶

Delete or move one cached model artifact directory to trash.

Source code in app/core/registry.py

def delete_cached_model(model_hash: str, registry_base: str | Path, soft_delete: bool = True) -> bool:
    """Delete or move one cached model artifact directory to trash."""
    if not isinstance(model_hash, str) or len(model_hash) < 4:
        return False
    base_path = Path(registry_base).resolve()
    target_dir = (base_path / model_hash).resolve()
    if (
        not base_path.exists()
        or not target_dir.is_relative_to(base_path)
        or target_dir == base_path
        or target_dir.name == ".trash"
        or not target_dir.is_dir()
    ):
        return False
    if soft_delete:
        trash_dir = base_path / ".trash"
        trash_dir.mkdir(parents=True, exist_ok=True)
        destination = trash_dir / model_hash
        if destination.exists():
            logger.warning("Refusing to overwrite existing trash entry: %s", destination)
            return False
        shutil.move(str(target_dir), str(destination))
        logger.info("Moved cached artifacts to trash: %s -> %s", target_dir, destination)
    else:
        shutil.rmtree(target_dir)
        logger.info("Permanently deleted cached artifacts: %s", target_dir)
    return True

`extract_and_save_pr_metrics(engine: Engine, model: Any, validation_dataloader: Any, test_dataloader: Any, base_dir: Path, run_heatmap: bool = False, model_name: str = 'PatchCore') -> EvaluationArtifacts` ¶

Extract model predictions and persist Precision-Recall metrics for visual analysis.

Parameters:

Name	Type	Description	Default
`engine`	`Engine`	Anomalib engine instance.	required
`model`	`Any`	Trained model.	required
`validation_dataloader`	`Any`	Loader containing only shared normal validation images.	required
`test_dataloader`	`Any`	Loader containing the unchanged official test partition.	required
`base_dir`	`Path`	Output directory for metrics.	required
`run_heatmap`	`bool`	Whether to compute heatmap overlays.	`False`
`model_name`	`str`	Human-readable model name used in diagnostics.	`'PatchCore'`

Returns:

Type	Description
`EvaluationArtifacts`	Structured EvaluationArtifacts container (also unpackable as 18-tuple for backward compatibility).

Source code in app/pipelines/modelling/patchcore/evaluation.py

def extract_and_save_pr_metrics(
    engine: Engine,
    model: Any,
    validation_dataloader: Any,
    test_dataloader: Any,
    base_dir: Path,
    run_heatmap: bool = False,
    model_name: str = "PatchCore",
) -> EvaluationArtifacts:
    """Extract model predictions and persist Precision-Recall metrics for visual analysis.

    Args:
        engine: Anomalib engine instance.
        model: Trained model.
        validation_dataloader: Loader containing only shared normal validation images.
        test_dataloader: Loader containing the unchanged official test partition.
        base_dir: Output directory for metrics.
        run_heatmap: Whether to compute heatmap overlays.
        model_name: Human-readable model name used in diagnostics.

    Returns:
        Structured EvaluationArtifacts container (also unpackable as 18-tuple for backward compatibility).
    """
    try:
        logger.info("Extracting predictions for PR curve metrics...")
        img_threshold, pix_threshold = _calibrate_validation_thresholds(
            engine, model, validation_dataloader, model_name
        )

        raw_predictions = engine.predict(model=model, dataloaders=test_dataloader)
        if not raw_predictions:
            raise RuntimeError(f"{model_name} prediction returned no test batches")

        anomaly_maps, ground_truth_masks, image_scores_np, image_labels_np = _collect_test_predictions(
            raw_predictions, model_name
        )

        image_metrics, pixel_metrics = _compute_and_persist_metrics(
            anomaly_maps,
            ground_truth_masks,
            image_scores_np,
            image_labels_np,
            img_threshold,
            pix_threshold,
            base_dir,
            model_name,
        )

        heatmap_overlays: dict[int, dict[str, list[Any]]] = {}
        anomalous_indices: list[int] = []
        if run_heatmap:
            heatmap_overlays, anomalous_indices = _generate_test_heatmaps(raw_predictions)

        return EvaluationArtifacts(
            image_metrics=image_metrics,
            pixel_metrics=pixel_metrics,
            thresholds=Thresholds(image=img_threshold, pixel=pix_threshold),
            heatmap_overlays=heatmap_overlays,
            anomalous_indices=anomalous_indices,
        )

    except Exception as e:
        logger.exception("Could not compute %s evaluation metrics: %s", model_name, e)
        raise

`find_cached_patchcore_model(category: str, backbone: str = 'resnet18', feature_layers: tuple[str, ...] = ('layer2', 'layer3'), coreset_sampling_ratio: float = 0.1, num_neighbors: int = 9, fpr_limit: float = 0.0001, pipeline: list[dict[str, Any]] | None = None, target_hash: str | None = None, registry_base: Path | str = 'data/models/patchcore', expected_split_evidence: dict[str, Any] | None = None) -> tuple[Path, dict[str, Any]] | None` ¶

Find the newest cached PatchCore model matching either a specific hash or the given parameters.

Parameters:

Name	Type	Description	Default
`category`	`str`	Component category name.	required
`backbone`	`str`	Feature extractor backbone name.	`'resnet18'`
`feature_layers`	`tuple[str, ...]`	Layers to extract features from.	`('layer2', 'layer3')`
`coreset_sampling_ratio`	`float`	Ratio for coreset subsampling.	`0.1`
`num_neighbors`	`int`	Number of nearest neighbors for scoring.	`9`
`fpr_limit`	`float`	Max allowable False Positive Rate.	`0.0001`
`pipeline`	`list[dict[str, Any]] \| None`	Optional preprocessing pipeline configuration.	`None`
`target_hash`	`str \| None`	Optional exact model hash to search for.	`None`
`registry_base`	`Path \| str`	Path to the patchcore model registry.	`'data/models/patchcore'`
`expected_split_evidence`	`dict[str, Any] \| None`	Required fair-protocol split evidence, when evaluating a cache hit.	`None`

Returns:

Type	Description
`tuple[Path, dict[str, Any]] \| None`	Tuple of (model_dir, metadata_dict) if found, else None.

Source code in app/pipelines/modelling/patchcore/registry.py

def find_cached_patchcore_model(
    category: str,
    backbone: str = "resnet18",
    feature_layers: tuple[str, ...] = ("layer2", "layer3"),
    coreset_sampling_ratio: float = 0.1,
    num_neighbors: int = 9,
    fpr_limit: float = 1e-4,
    pipeline: list[dict[str, Any]] | None = None,
    target_hash: str | None = None,
    registry_base: Path | str = "data/models/patchcore",
    expected_split_evidence: dict[str, Any] | None = None,
) -> tuple[Path, dict[str, Any]] | None:
    """Find the newest cached PatchCore model matching either a specific hash or the given parameters.

    Args:
        category: Component category name.
        backbone: Feature extractor backbone name.
        feature_layers: Layers to extract features from.
        coreset_sampling_ratio: Ratio for coreset subsampling.
        num_neighbors: Number of nearest neighbors for scoring.
        fpr_limit: Max allowable False Positive Rate.
        pipeline: Optional preprocessing pipeline configuration.
        target_hash: Optional exact model hash to search for.
        registry_base: Path to the patchcore model registry.
        expected_split_evidence: Required fair-protocol split evidence, when evaluating a cache hit.

    Returns:
        Tuple of (model_dir, metadata_dict) if found, else None.
    """
    base_path = Path(registry_base)
    if not base_path.exists():
        return None

    if target_hash:
        return _find_target_hash_cached_dir(base_path, target_hash, expected_split_evidence)

    if isinstance(pipeline, PreprocessingPipeline):
        norm_req_prep: list[dict[str, Any]] = []
    else:
        norm_req_prep = _normalize_preprocessing_steps(pipeline)
    candidates: list[tuple[float, Path, dict[str, Any]]] = []

    for meta_file in base_path.rglob("metadata.json"):
        if ".trash" in meta_file.parts:
            continue
        try:
            with open(meta_file, encoding="utf-8") as f:
                meta = json.load(f)
        except Exception:
            continue

        if not _matches_patchcore_metadata(
            meta=meta,
            category=category,
            backbone=backbone,
            feature_layers=feature_layers,
            coreset_sampling_ratio=coreset_sampling_ratio,
            num_neighbors=num_neighbors,
            fpr_limit=fpr_limit,
            norm_req_prep=norm_req_prep,
            expected_split_evidence=expected_split_evidence,
        ):
            continue

        ts = _extract_metadata_timestamp(meta, meta_file)
        candidates.append((ts, meta_file.parent, meta))

    if not candidates:
        return None

    candidates.sort(key=lambda x: x[0], reverse=True)
    _, newest_dir, newest_meta = candidates[0]
    return newest_dir, newest_meta

format_results(test_results: list[Mapping[str, float]] | None, category: str, base_dir: Path, manual_image_f1: float = 0.0, manual_pixel_f1: float = 0.0, manual_image_prec: float = 0.0, manual_image_rec: float = 0.0, img_threshold: float = 0.0, pixel_threshold: float = 0.0, pixel_auroc: float = 0.0, pixel_aupimo: float = 0.0, anomaly_map_min: float = 0.0, anomaly_map_max: float = 0.0, anomaly_map_range: float = 0.0, heatmap_overlays: dict[int, dict[str, list[Any]]] | None = None, anomalous_indices: list[int] | None = None, fpr_limit: float = 0.0001, preprocessing_steps: list[dict[str, Any]] | None = None, hyperparameters: dict[str, Any] | None = None, dataset_split: dict[str, Any] | None = None, model_hash: str = '', metadata: dict[str, Any] | None = None, true_positives: int = 0, false_positives: int = 0, false_negatives: int = 0, true_negatives: int = 0, artifacts: EvaluationArtifacts | None = None) -> BaselineResult ¶

Format Anomalib engine evaluation output into a structured response schema.

Parameters:

Name	Type	Description	Default
`test_results`	`list[Mapping[str, float]] \| None`	A list of metric mappings from Anomalib.	required
`category`	`str`	The component category name.	required
`base_dir`	`Path`	Base directory to save metrics to.	required
`manual_image_f1`	`float`	Manually calculated image-level F1 score.	`0.0`
`manual_pixel_f1`	`float`	Manually calculated pixel-level F1 score.	`0.0`
`manual_image_prec`	`float`	Manually calculated image-level Precision score.	`0.0`
`manual_image_rec`	`float`	Manually calculated image-level Recall score.	`0.0`
`img_threshold`	`float`	Manually calculated image-level classification threshold.	`0.0`
`pixel_threshold`	`float`	Normal-validation threshold used to create predicted masks.	`0.0`
`pixel_auroc`	`float`	Pixel AUROC from the shared canonical metric path.	`0.0`
`pixel_aupimo`	`float`	Full-map AUPIMO computed by Anomalib.	`0.0`
`anomaly_map_min`	`float`	Minimum PatchCore anomaly-map value.	`0.0`
`anomaly_map_max`	`float`	Maximum PatchCore anomaly-map value.	`0.0`
`anomaly_map_range`	`float`	Range of PatchCore anomaly-map values.	`0.0`
`heatmap_overlays`	`dict[int, dict[str, list[Any]]] \| None`	Dictionary of precomputed heatmap overlays.	`None`
`anomalous_indices`	`list[int] \| None`	List of image indices corresponding to anomalies.	`None`
`fpr_limit`	`float`	Maximum allowable False Positive Rate for AUPIMO threshold.	`0.0001`
`preprocessing_steps`	`list[dict[str, Any]] \| None`	Optional list of active preprocessing steps.	`None`
`hyperparameters`	`dict[str, Any] \| None`	Optional dictionary of model hyperparameters.	`None`
`dataset_split`	`dict[str, Any] \| None`	Optional dataset partition sample counts.	`None`
`model_hash`	`str`	Unique 12-char model hash.	`''`
`metadata`	`dict[str, Any] \| None`	Full metadata dictionary.	`None`
`true_positives`	`int`	Image-level true-positive count.	`0`
`false_positives`	`int`	Image-level false-positive count.	`0`
`false_negatives`	`int`	Image-level false-negative count.	`0`
`true_negatives`	`int`	Image-level true-negative count.	`0`
`artifacts`	`EvaluationArtifacts \| None`	Optional strongly-typed EvaluationArtifacts container. If passed, individual metric values are automatically derived from it.	`None`

Returns:

Type	Description
`BaselineResult`	A dictionary containing structured image_level and pixel_level results.

Source code in app/pipelines/modelling/patchcore/evaluation.py

def format_results(
    test_results: list[Mapping[str, float]] | None,
    category: str,
    base_dir: Path,
    manual_image_f1: float = 0.0,
    manual_pixel_f1: float = 0.0,
    manual_image_prec: float = 0.0,
    manual_image_rec: float = 0.0,
    img_threshold: float = 0.0,
    pixel_threshold: float = 0.0,
    pixel_auroc: float = 0.0,
    pixel_aupimo: float = 0.0,
    anomaly_map_min: float = 0.0,
    anomaly_map_max: float = 0.0,
    anomaly_map_range: float = 0.0,
    heatmap_overlays: dict[int, dict[str, list[Any]]] | None = None,
    anomalous_indices: list[int] | None = None,
    fpr_limit: float = 1e-4,
    preprocessing_steps: list[dict[str, Any]] | None = None,
    hyperparameters: dict[str, Any] | None = None,
    dataset_split: dict[str, Any] | None = None,
    model_hash: str = "",
    metadata: dict[str, Any] | None = None,
    true_positives: int = 0,
    false_positives: int = 0,
    false_negatives: int = 0,
    true_negatives: int = 0,
    artifacts: EvaluationArtifacts | None = None,
) -> BaselineResult:
    """Format Anomalib engine evaluation output into a structured response schema.

    Args:
        test_results: A list of metric mappings from Anomalib.
        category: The component category name.
        base_dir: Base directory to save metrics to.
        manual_image_f1: Manually calculated image-level F1 score.
        manual_pixel_f1: Manually calculated pixel-level F1 score.
        manual_image_prec: Manually calculated image-level Precision score.
        manual_image_rec: Manually calculated image-level Recall score.
        img_threshold: Manually calculated image-level classification threshold.
        pixel_threshold: Normal-validation threshold used to create predicted masks.
        pixel_auroc: Pixel AUROC from the shared canonical metric path.
        pixel_aupimo: Full-map AUPIMO computed by Anomalib.
        anomaly_map_min: Minimum PatchCore anomaly-map value.
        anomaly_map_max: Maximum PatchCore anomaly-map value.
        anomaly_map_range: Range of PatchCore anomaly-map values.
        heatmap_overlays: Dictionary of precomputed heatmap overlays.
        anomalous_indices: List of image indices corresponding to anomalies.
        fpr_limit: Maximum allowable False Positive Rate for AUPIMO threshold.
        preprocessing_steps: Optional list of active preprocessing steps.
        hyperparameters: Optional dictionary of model hyperparameters.
        dataset_split: Optional dataset partition sample counts.
        model_hash: Unique 12-char model hash.
        metadata: Full metadata dictionary.
        true_positives: Image-level true-positive count.
        false_positives: Image-level false-positive count.
        false_negatives: Image-level false-negative count.
        true_negatives: Image-level true-negative count.
        artifacts: Optional strongly-typed EvaluationArtifacts container. If passed,
            individual metric values are automatically derived from it.

    Returns:
        A dictionary containing structured image_level and pixel_level results.
    """
    if not np.isclose(fpr_limit, AUPIMO_FPR_BOUNDS[1]):
        raise ValueError(f"fair-eval-v1 requires fpr_limit={AUPIMO_FPR_BOUNDS[1]}")

    if artifacts is not None:
        artifacts = EvaluationArtifacts.from_tuple(artifacts)
        manual_image_f1 = artifacts.image_metrics.f1_score
        manual_pixel_f1 = artifacts.pixel_metrics.f1_score
        manual_image_prec = artifacts.image_metrics.precision
        manual_image_rec = artifacts.image_metrics.recall
        img_threshold = artifacts.thresholds.image
        pixel_threshold = artifacts.thresholds.pixel
        pixel_auroc = artifacts.pixel_metrics.auroc
        pixel_aupimo = artifacts.pixel_metrics.aupimo
        anomaly_map_min = artifacts.pixel_metrics.anomaly_map_min
        anomaly_map_max = artifacts.pixel_metrics.anomaly_map_max
        anomaly_map_range = artifacts.pixel_metrics.anomaly_map_range
        heatmap_overlays = artifacts.heatmap_overlays
        anomalous_indices = artifacts.anomalous_indices
        true_positives = artifacts.image_metrics.confusion.true_positives
        false_positives = artifacts.image_metrics.confusion.false_positives
        false_negatives = artifacts.image_metrics.confusion.false_negatives
        true_negatives = artifacts.image_metrics.confusion.true_negatives

    res_dict: Mapping[str, float] = test_results[0] if test_results else {}

    return {
        "category": category,
        "image_level": {
            "auroc": _to_float(res_dict.get("image_AUROC", 0.0)),
            "f1_score": manual_image_f1,
            "precision": manual_image_prec,
            "recall": manual_image_rec,
            "threshold": img_threshold,
            "true_positives": true_positives,
            "false_positives": false_positives,
            "false_negatives": false_negatives,
            "true_negatives": true_negatives,
            "metrics_path": str(base_dir / "image_metrics.npz"),
        },
        "pixel_level": {
            "auroc": pixel_auroc,
            "f1_score": manual_pixel_f1,
            "threshold": pixel_threshold,
            "aupimo_score": pixel_aupimo,
            "fpr_lower_bound": 1e-5,
            "fpr_upper_bound": fpr_limit,
            "aupimo_num_thresholds": AUPIMO_NUM_THRESHOLDS,
            "canonical_height": CANONICAL_MAP_SIZE[0],
            "canonical_width": CANONICAL_MAP_SIZE[1],
            "aupimo": pixel_aupimo,
            "anomaly_map_min": anomaly_map_min,
            "anomaly_map_max": anomaly_map_max,
            "anomaly_map_range": anomaly_map_range,
            "metrics_path": str(base_dir / "pixel_metrics.npz"),
        },
        "raw_results": {k: _to_float(v) for k, v in res_dict.items()},
        "heatmap_overlays": heatmap_overlays or {},
        "anomalous_indices": anomalous_indices or [],
        "preprocessing_steps": preprocessing_steps or [],
        "hyperparameters": hyperparameters or {},
        "dataset_split": dataset_split or {},
        "model_hash": model_hash,
        "metadata": metadata or {},
    }

`list_trashed_patchcore_models(registry_base: str | Path) -> list[dict[str, Any]]` ¶

Return metadata for all artifact directories in trash.

Source code in app/core/registry.py

def list_trashed_models(registry_base: str | Path) -> list[dict[str, Any]]:
    """Return metadata for all artifact directories in trash."""
    trash_dir = Path(registry_base).resolve() / ".trash"
    if not trash_dir.exists():
        return []
    trashed: list[dict[str, Any]] = []
    for metadata_path in trash_dir.rglob("metadata.json"):
        try:
            metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
            metadata["hash"] = metadata.get("hash", metadata_path.parent.name)
            trashed.append(metadata)
        except (OSError, json.JSONDecodeError):
            trashed.append({"hash": metadata_path.parent.name})
    return trashed

`purge_patchcore_trash(registry_base: str | Path, model_hash: str | None = None) -> int` ¶

Permanently remove cached artifact directorie(s) in trash.

Parameters:

Name	Type	Description	Default
`registry_base`	`str \| Path`	Base directory for the registry.	required
`model_hash`	`str \| None`	Optional specific model hash to purge. If None, all trash is purged.	`None`

Source code in app/core/registry.py

def purge_trash(registry_base: str | Path, model_hash: str | None = None) -> int:
    """Permanently remove cached artifact directorie(s) in trash.

    Args:
        registry_base: Base directory for the registry.
        model_hash: Optional specific model hash to purge. If None, all trash is purged.
    """
    trash_dir = Path(registry_base).resolve() / ".trash"
    if not trash_dir.exists():
        return 0
    purged = 0
    if model_hash is not None:
        target = trash_dir / model_hash
        if target.exists() and target.is_dir():
            shutil.rmtree(target)
            purged += 1
            logger.info("Purged specific cached run from trash: %s", model_hash)
    else:
        for child in trash_dir.iterdir():
            if child.is_dir():
                shutil.rmtree(child)
                purged += 1
        logger.info("Emptied trash: purged %d cached run(s).", purged)
    return purged

`restore_cached_patchcore_model(model_hash: str, registry_base: str | Path) -> bool` ¶

Restore one soft-deleted artifact directory.

Source code in app/core/registry.py

def restore_cached_model(model_hash: str, registry_base: str | Path) -> bool:
    """Restore one soft-deleted artifact directory."""
    if not isinstance(model_hash, str) or len(model_hash) < 4:
        return False
    base_path = Path(registry_base).resolve()
    source = (base_path / ".trash" / model_hash).resolve()
    destination = (base_path / model_hash).resolve()
    if not source.is_relative_to(base_path / ".trash") or not source.is_dir():
        return False
    if destination.exists():
        logger.warning("Refusing to overwrite active run while restoring: %s", destination)
        return False
    shutil.move(str(source), str(destination))
    logger.info("Restored cached artifacts from trash: %s -> %s", source, destination)
    return True

run_patchcore_pipeline(data_root: Path | str = 'data/raw/mvtec_ad', category: str = 'bottle', pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None, fpr_limit: float = 0.0001, backbone: str = 'resnet18', feature_layers: tuple[str, ...] = ('layer2', 'layer3'), coreset_sampling_ratio: float = 0.1, num_neighbors: int = 9, run_heatmap: bool = False, force_retrain: bool = False, model_hash: str | None = None, registry_base: Path | str = 'data/models/patchcore', model_seed: int = PATCHCORE_MODEL_SEED) -> BaselineResult ¶

Run the PatchCore anomaly detection pipeline on the MVTec AD dataset.

Parameters:

Name	Type	Description	Default
`data_root`	`Path \| str`	Root directory of MVTec AD.	`'data/raw/mvtec_ad'`
`category`	`str`	Category to evaluate.	`'bottle'`
`pipeline`	`list[dict[str, Any]] \| PreprocessingPipeline \| None`	Optional list of preprocessing step configurations or pipeline.	`None`
`fpr_limit`	`float`	Maximum allowable False Positive Rate.	`0.0001`
`backbone`	`str`	Feature extractor backbone (e.g. 'resnet18', 'wide_resnet50_2').	`'resnet18'`
`feature_layers`	`tuple[str, ...]`	Layers to extract features from.	`('layer2', 'layer3')`
`coreset_sampling_ratio`	`float`	Ratio for coreset subsampling.	`0.1`
`num_neighbors`	`int`	Number of nearest neighbors for scoring.	`9`
`run_heatmap`	`bool`	Whether to compute heatmap overlays.	`False`
`force_retrain`	`bool`	If True, ignores cache and forces a full re-fit.	`False`
`model_hash`	`str \| None`	Optional target model hash to search for.	`None`
`registry_base`	`Path \| str`	Base directory path for Patchcore model registry.	`'data/models/patchcore'`
`model_seed`	`int`	Seed controlling PatchCore coreset sampling and data-loader workers.	`PATCHCORE_MODEL_SEED`

Returns:

Type	Description
`BaselineResult`	Structured evaluation metrics conforming to fair-eval-v1.

Source code in app/pipelines/modelling/patchcore/pipeline.py

def run_patchcore_pipeline(
    data_root: Path | str = "data/raw/mvtec_ad",
    category: str = "bottle",
    pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None,
    fpr_limit: float = 1e-4,
    backbone: str = "resnet18",
    feature_layers: tuple[str, ...] = ("layer2", "layer3"),
    coreset_sampling_ratio: float = 0.1,
    num_neighbors: int = 9,
    run_heatmap: bool = False,
    force_retrain: bool = False,
    model_hash: str | None = None,
    registry_base: Path | str = "data/models/patchcore",
    model_seed: int = PATCHCORE_MODEL_SEED,
) -> BaselineResult:
    """Run the PatchCore anomaly detection pipeline on the MVTec AD dataset.

    Args:
        data_root: Root directory of MVTec AD.
        category: Category to evaluate.
        pipeline: Optional list of preprocessing step configurations or pipeline.
        fpr_limit: Maximum allowable False Positive Rate.
        backbone: Feature extractor backbone (e.g. 'resnet18', 'wide_resnet50_2').
        feature_layers: Layers to extract features from.
        coreset_sampling_ratio: Ratio for coreset subsampling.
        num_neighbors: Number of nearest neighbors for scoring.
        run_heatmap: Whether to compute heatmap overlays.
        force_retrain: If True, ignores cache and forces a full re-fit.
        model_hash: Optional target model hash to search for.
        registry_base: Base directory path for Patchcore model registry.
        model_seed: Seed controlling PatchCore coreset sampling and data-loader workers.

    Returns:
        Structured evaluation metrics conforming to fair-eval-v1.
    """
    if not np.isclose(fpr_limit, AUPIMO_FPR_BOUNDS[1]):
        raise ValueError(f"fair-eval-v1 requires fpr_limit={AUPIMO_FPR_BOUNDS[1]}")

    if isinstance(pipeline, PreprocessingPipeline):
        proc_pipeline = pipeline
        raw_prep_list: list[dict[str, Any]] = []
    else:
        proc_pipeline = build_pipeline_from_configs(pipeline)
        raw_prep_list = _normalize_preprocessing_steps(pipeline)

    manifest = build_mvtec_manifest(data_root)
    fair_split = build_fair_evaluation_split(manifest, category)
    split_evidence = fair_split.evidence()
    cache_evidence = {
        **split_evidence,
        **fair_metric_evidence(),
        "model_seed": model_seed,
        "score_space": PATCHCORE_SCORE_SPACE,
        "image_threshold_quantile": PATCHCORE_IMAGE_THRESHOLD_QUANTILE,
        "pixel_threshold_quantile": PATCHCORE_PIXEL_THRESHOLD_QUANTILE,
    }
    norm_prep_str = json.dumps(raw_prep_list, sort_keys=True)
    layer_str = "_".join(feature_layers)
    hp_string = (
        f"{category}_{backbone}_{layer_str}_{coreset_sampling_ratio}_{num_neighbors}_{fpr_limit}_{norm_prep_str}_"
        f"{json.dumps(cache_evidence, sort_keys=True)}_{CANONICAL_MAP_SIZE}_{AUPIMO_FPR_BOUNDS}_"
        f"{AUPIMO_NUM_THRESHOLDS}_{PIXEL_METRICS_VERSION}"
    )
    computed_hash = hashlib.sha256(hp_string.encode()).hexdigest()[:12]

    cached = find_cached_patchcore_model(
        category=category,
        backbone=backbone,
        feature_layers=feature_layers,
        coreset_sampling_ratio=coreset_sampling_ratio,
        num_neighbors=num_neighbors,
        fpr_limit=fpr_limit,
        pipeline=raw_prep_list,
        target_hash=model_hash,
        registry_base=registry_base,
        expected_split_evidence=cache_evidence,
    )
    if model_hash and cached is None:
        raise FileNotFoundError(
            f"Cached PatchCore run {model_hash} is missing or does not match the current dataset protocol."
        )

    if cached is not None and not force_retrain:
        cached_dir, meta = cached
        return _load_cached_patchcore_result(
            cached_dir=cached_dir,
            meta=meta,
            category=category,
            backbone=backbone,
            feature_layers=feature_layers,
            coreset_sampling_ratio=coreset_sampling_ratio,
            num_neighbors=num_neighbors,
            fpr_limit=fpr_limit,
            raw_prep_list=raw_prep_list,
        )

    # A rejected or explicitly bypassed cache must never be overwritten.
    effective_hash = computed_hash
    logger.info("Configured preprocessing pipeline with %d steps.", len(proc_pipeline))
    base_dir = Path(registry_base) / effective_hash
    if base_dir.exists():
        archived_hash = hashlib.sha256(f"{computed_hash}:{time.time_ns()}".encode()).hexdigest()[:12]
        archived_dir = Path(registry_base) / archived_hash
        base_dir.rename(archived_dir)
        archived_metadata_path = archived_dir / "metadata.json"
        if archived_metadata_path.is_file():
            archived_metadata = json.loads(archived_metadata_path.read_text(encoding="utf-8"))
            archived_metadata["hash"] = archived_hash
            archived_metadata["configuration_hash"] = computed_hash
            archived_metadata_path.write_text(json.dumps(archived_metadata, indent=4), encoding="utf-8")
    base_dir.mkdir(parents=True)
    transform_adapter = PreprocessingTransformAdapter(proc_pipeline)

    # 1. Initialize dataset, model, and engine
    datamodule = MVTecAD(
        root=data_root,
        category=category,
        train_batch_size=16,
        eval_batch_size=16,
        val_split_mode="none",
    )
    _configure_patchcore_partitions(
        datamodule,
        fair_split,
        transform_adapter if len(proc_pipeline) > 0 else None,
    )

    # PatchCore's coreset is sampled stochastically. Seed immediately before construction.
    _seed_patchcore_run(model_seed)
    model = Patchcore(
        backbone=backbone,
        layers=feature_layers,
        coreset_sampling_ratio=coreset_sampling_ratio,
        num_neighbors=num_neighbors,
        post_processor=False,
        evaluator=False,
        visualizer=_RawScoreImageVisualizer(output_dir=base_dir / "four_panel_images"),
    )
    engine = Engine(accelerator="gpu", devices=1, deterministic=True)

    # 2. Fit
    train_dataloader = datamodule.train_dataloader()
    validation_dataloader = datamodule.val_dataloader()
    test_dataloader = datamodule.test_dataloader()
    logger.info("Fitting Patchcore model on %s category (Hash: %s)...", category, effective_hash)
    engine.fit(model, train_dataloaders=train_dataloader)

    # 3. Extract PR metrics and build summary
    artifacts = EvaluationArtifacts.from_tuple(
        extract_and_save_pr_metrics(
            engine,
            model,
            validation_dataloader,
            test_dataloader,
            base_dir,
            run_heatmap,
        )
    )

    # Extract dataset split counts
    split_info = {
        **cache_evidence,
        "test_normal": (fair_split.test["is_anomaly"] == 0).sum(),
        "test_anomalous": fair_split.test["is_anomaly"].astype(bool).sum(),
    }

    hyperparams = {
        "backbone": backbone,
        "feature_layers": feature_layers,
        "coreset_sampling_ratio": coreset_sampling_ratio,
        "num_neighbors": num_neighbors,
        "fpr_limit": fpr_limit,
        "train_batch_size": 16,
        "eval_batch_size": 16,
        "model_seed": model_seed,
        "score_space": PATCHCORE_SCORE_SPACE,
    }

    raw_results_dict = {
        "image_AUROC": artifacts.image_metrics.auroc,
        "image_F1Score": artifacts.image_metrics.f1_score,
        "image_Precision": artifacts.image_metrics.precision,
        "image_Recall": artifacts.image_metrics.recall,
        "pixel_AUROC": artifacts.pixel_metrics.auroc,
        "pixel_F1Score": artifacts.pixel_metrics.f1_score,
        "pixel_AUPIMO": artifacts.pixel_metrics.aupimo,
    }
    test_results: list[Mapping[str, float]] = [raw_results_dict]
    _print_patchcore_results_table(raw_results_dict)

    logger.info(
        "PatchCore evaluation summary | image: AUROC=%.6f F1=%.6f precision=%.6f recall=%.6f threshold=%.6f",
        artifacts.image_metrics.auroc,
        artifacts.image_metrics.f1_score,
        artifacts.image_metrics.precision,
        artifacts.image_metrics.recall,
        artifacts.thresholds.image,
    )
    logger.info(
        "PatchCore evaluation summary | confusion: TP=%d FP=%d FN=%d TN=%d",
        artifacts.image_metrics.confusion.true_positives,
        artifacts.image_metrics.confusion.false_positives,
        artifacts.image_metrics.confusion.false_negatives,
        artifacts.image_metrics.confusion.true_negatives,
    )
    logger.info(
        "PatchCore evaluation summary | pixel: AUROC=%.6f F1=%.6f AUPIMO=%.6f",
        artifacts.pixel_metrics.auroc,
        artifacts.pixel_metrics.f1_score,
        artifacts.pixel_metrics.aupimo,
    )

    heatmap_archive = _save_heatmap_overlays(artifacts.heatmap_overlays, base_dir / "heatmap_overlays.npz")
    four_panel_dir = base_dir / "four_panel_images"
    if heatmap_archive is not None:
        logger.info("Saved compressed PatchCore heatmaps to %s", heatmap_archive)

    metadata = {
        "hash": effective_hash,
        "configuration_hash": computed_hash,
        "model_type": "patchcore",
        "category": category,
        "backbone": backbone,
        "feature_layers": list(feature_layers),
        "coreset_sampling_ratio": coreset_sampling_ratio,
        "num_neighbors": num_neighbors,
        "fpr_limit": fpr_limit,
        "preprocessing_steps": raw_prep_list,
        "hyperparameters": hyperparams,
        "dataset_split": split_info,
        "protocol": FAIR_EVALUATION_PROTOCOL,
        "threshold_source": "normal_validation",
        "image_threshold_quantile": PATCHCORE_IMAGE_THRESHOLD_QUANTILE,
        "pixel_threshold_quantile": PATCHCORE_PIXEL_THRESHOLD_QUANTILE,
        "model_seed": model_seed,
        "score_space": PATCHCORE_SCORE_SPACE,
        "pixel_metrics_version": PIXEL_METRICS_VERSION,
        "image_auroc": raw_results_dict["image_AUROC"],
        "pixel_auroc": artifacts.pixel_metrics.auroc,
        "manual_image_f1": artifacts.image_metrics.f1_score,
        "manual_pixel_f1": artifacts.pixel_metrics.f1_score,
        "manual_image_prec": artifacts.image_metrics.precision,
        "manual_image_rec": artifacts.image_metrics.recall,
        "true_positives": artifacts.image_metrics.confusion.true_positives,
        "false_positives": artifacts.image_metrics.confusion.false_positives,
        "false_negatives": artifacts.image_metrics.confusion.false_negatives,
        "true_negatives": artifacts.image_metrics.confusion.true_negatives,
        "img_threshold": artifacts.thresholds.image,
        "pixel_threshold": artifacts.thresholds.pixel,
        "pixel_aupimo": artifacts.pixel_metrics.aupimo,
        "aupimo_fpr_bounds": [1e-5, fpr_limit],
        "aupimo_num_thresholds": AUPIMO_NUM_THRESHOLDS,
        "canonical_height": CANONICAL_MAP_SIZE[0],
        "canonical_width": CANONICAL_MAP_SIZE[1],
        "anomaly_map_min": artifacts.pixel_metrics.anomaly_map_min,
        "anomaly_map_max": artifacts.pixel_metrics.anomaly_map_max,
        "anomaly_map_range": artifacts.pixel_metrics.anomaly_map_range,
        "heatmap_overlays_path": heatmap_archive.name if heatmap_archive is not None else None,
        "four_panel_images_path": four_panel_dir.name if four_panel_dir.is_dir() else None,
        "anomalous_indices": artifacts.anomalous_indices,
        "raw_results": raw_results_dict,
        "timestamp": datetime.now(UTC).isoformat(),
    }

    try:
        with open(base_dir / "metadata.json", "w", encoding="utf-8") as f:
            json.dump(metadata, f, indent=4)
        logger.info("Saved Patchcore model metadata to %s", base_dir / "metadata.json")
    except Exception as e:
        logger.warning("Could not save Patchcore metadata.json: %s", e)

    return format_results(
        test_results=test_results,
        category=category,
        base_dir=base_dir,
        artifacts=artifacts,
        fpr_limit=fpr_limit,
        preprocessing_steps=raw_prep_list,
        hyperparameters=hyperparams,
        dataset_split=split_info,
        model_hash=effective_hash,
        metadata=metadata,
    )

`app.pipelines.modelling.patchcore.optuna_study` ¶

Category-Adaptive Optuna Optimization for PatchCore.

Performs Bayesian hyperparameter sweeps across backbone architectures, feature extraction layers, coreset sampling ratios, and domain preprocessing.

`objective(trial: optuna.Trial, category_name: str, data_root: str = 'data/raw/mvtec_ad') -> float` ¶

Optuna objective function for tuning PatchCore hyperparameters.

Parameters:

Name	Type	Description	Default
`trial`	`Trial`	Active Optuna trial instance.	required
`category_name`	`str`	MVTec AD category string.	required
`data_root`	`str`	Root dataset folder path.	`'data/raw/mvtec_ad'`

Returns:

Type	Description
`float`	Objective evaluation metric score for the trial.

Source code in app/pipelines/modelling/patchcore/optuna_study.py

def objective(trial: optuna.Trial, category_name: str, data_root: str = "data/raw/mvtec_ad") -> float:
    """Optuna objective function for tuning PatchCore hyperparameters.

    Args:
        trial: Active Optuna trial instance.
        category_name: MVTec AD category string.
        data_root: Root dataset folder path.

    Returns:
        Objective evaluation metric score for the trial.
    """
    is_texture = category_name in TEXTURES
    backbone = "resnet18"

    layer_config = trial.suggest_categorical("feature_layers", ["l2_l3", "l2_l3_l4"])
    feature_layers: tuple[str, ...]
    if layer_config == "l2_l3":
        feature_layers = ("layer2", "layer3")
    else:
        feature_layers = ("layer2", "layer3", "layer4")

    coreset_ratio = trial.suggest_float("coreset_sampling_ratio", 0.001, 0.20, log=True)
    num_neighbors = trial.suggest_int("num_neighbors", 1, 9)

    use_clahe = trial.suggest_categorical("use_clahe", [True, False])
    use_gaussian_blur = trial.suggest_categorical("use_gaussian_blur", [True, False])

    use_foreground_mask = False if is_texture else trial.suggest_categorical("use_foreground_mask", [True, False])

    return _evaluate_patchcore(
        category_name=category_name,
        backbone=backbone,
        feature_layers=feature_layers,
        coreset_ratio=coreset_ratio,
        num_neighbors=num_neighbors,
        use_clahe=use_clahe,
        use_gaussian_blur=use_gaussian_blur,
        use_foreground_mask=use_foreground_mask,
        data_root=data_root,
    )

`run_study(category_name: str, n_trials: int = 30, data_root: str = 'data/raw/mvtec_ad') -> dict[str, Any]` ¶

Execute Optuna optimization study for PatchCore on a specific category.

Parameters:

Name	Type	Description	Default
`category_name`	`str`	Target category to tune.	required
`n_trials`	`int`	Maximum number of trials to evaluate.	`30`
`data_root`	`str`	Dataset root directory path.	`'data/raw/mvtec_ad'`

Returns:

Type	Description
`dict[str, Any]`	Structured configuration dictionary of the best hyperparameter settings found.

Source code in app/pipelines/modelling/patchcore/optuna_study.py

def run_study(category_name: str, n_trials: int = 30, data_root: str = "data/raw/mvtec_ad") -> dict[str, Any]:
    """Execute Optuna optimization study for PatchCore on a specific category.

    Args:
        category_name: Target category to tune.
        n_trials: Maximum number of trials to evaluate.
        data_root: Dataset root directory path.

    Returns:
        Structured configuration dictionary of the best hyperparameter settings found.
    """
    study_name = f"patchcore_{category_name}"

    storage_path = Path("data/hyperparameters/patchcore_optuna.db")
    storage_path.parent.mkdir(parents=True, exist_ok=True)
    storage_url = f"sqlite:///{storage_path.resolve()}"

    study = optuna.create_study(
        study_name=study_name,
        storage=storage_url,
        load_if_exists=True,
        direction="maximize",
        pruner=optuna.pruners.MedianPruner(),
    )

    trials_to_run = max(0, n_trials - len(study.trials))
    if trials_to_run > 0:
        study.optimize(lambda t: objective(t, category_name, data_root), n_trials=trials_to_run)

    best_trial = study.best_trial
    logger.info("Best trial for %s:", category_name)
    logger.info("  Value (Pixel AUPIMO): %f", best_trial.value)
    logger.info("  Params: ")
    for key, value in best_trial.params.items():
        logger.info("    %s: %s", key, value)

    is_texture = category_name in TEXTURES

    return {
        "target_metric": "pixel_auroc",
        "score": best_trial.value,
        "preprocessing": {
            "use_foreground_mask": False if is_texture else best_trial.params.get("use_foreground_mask", False),
            "use_clahe": best_trial.params["use_clahe"],
            "use_gaussian_blur": best_trial.params["use_gaussian_blur"],
        },
        "model_hyperparameters": {
            "backbone": "resnet18",
            "feature_layers": best_trial.params["feature_layers"],
            "coreset_sampling_ratio": best_trial.params["coreset_sampling_ratio"],
            "num_neighbors": best_trial.params["num_neighbors"],
        },
    }

`app.pipelines.modelling.keras_cae.cae_pipeline` ¶

End-to-end orchestrator for the Keras Convolutional Autoencoder (CAE) pipeline.

Coordinates the complete anomaly detection workflow for MVTec AD categories under the deterministic fair-eval-v1 evaluation protocol: 1. Data Loading & Partitioning: Loads dataset manifests and partitions normal samples into 85% fit and 15% validation subsets with zero test leakage. 2. Preprocessing & Patching: Applies optional filters (CLAHE, blur, foreground masks) and extracts sliding-window crops. 3. CAE Modeling: Builds and trains a convolutional autoencoder using Masked Image Modeling (MIM) with joint SSIM and MSE reconstruction loss. 4. Scoring & Calibration: Generates pixel error maps, aggregates image scores via Top-K spatial pooling, and calibrates decision thresholds strictly on normal validation data. 5. Evaluation & Persistence: Computes canonical image AUROC, strict AUPIMO, confusion matrices, and heatmaps, with deterministic caching and soft-delete trash management.

run_keras_cae_pipeline(data_root: str = 'data/raw/mvtec_ad', category: str = 'bottle', img_size: int = 256, crop_size: int = 64, crop_stride: int = 32, latent_channels: int = 32, epochs: int = 20, batch_size: int = 16, mask_ratio: float = 0.25, mask_patch_size: int = 8, threshold_method: str = 'quantile', k_fraction: float = 0.002, pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None, run_heatmap: bool = False, force_retrain: bool = False, model_hash: str | None = None, trial: Any | None = None) -> dict[str, Any] ¶

Run the complete Keras CAE anomaly detection pipeline for one MVTec category.

This is the main entry point called by the Streamlit application. It: 1. Resolves cached model or configures new training parameters. 2. Loads train (normal only) and test images as numpy arrays using exact parameters. 3. Applies modular preprocessing transforms consistent with model state. 4. Normalises images to [0, 1]. 5. Builds and trains the Keras CAE with MIM + SSIM+MSE + AdamW (or loads from cache). 6. Scores all test images using Top-K pooling. 7. Computes an adaptive threshold from normal validation scores. 8. Evaluates with image-level AUROC and pixel-level AUPIMO. 9. Optionally computes Reconstruction Error Heatmap overlays for every anomalous test image.

Parameters:

Name	Type	Description	Default
`data_root`	`str`	Path to the MVTec AD dataset root directory.	`'data/raw/mvtec_ad'`
`category`	`str`	MVTec category to train and evaluate on (e.g., 'bottle', 'wood').	`'bottle'`
`img_size`	`int`	Size (height and width) to resize base images to.	`256`
`crop_size`	`int`	Size of the sliding window crops extracted from the base image.	`64`
`crop_stride`	`int`	Stride of the sliding window.	`32`
`latent_channels`	`int`	Number of channels in the convolutional bottleneck.	`32`
`epochs`	`int`	Number of training epochs.	`20`
`batch_size`	`int`	Training batch size (number of crops, not full images).	`16`
`mask_ratio`	`float`	Fraction of patches to mask during Masked Image Modeling training.	`0.25`
`mask_patch_size`	`int`	Side length of each masked region within a crop.	`8`
`threshold_method`	`str`	`"quantile"` or `"mahalanobis"` for adaptive threshold.	`'quantile'`
`k_fraction`	`float`	Top-K fraction for image-level anomaly score pooling.	`0.002`
`pipeline`	`list[dict[str, Any]] \| PreprocessingPipeline \| None`	Optional configuration list or PreprocessingPipeline object.	`None`
`run_heatmap`	`bool`	Whether to compute Reconstruction Error heatmap overlays for anomalous images.	`False`
`force_retrain`	`bool`	If True, bypass the cache and force training of a new model.	`False`
`model_hash`	`str \| None`	Optional specific model hash to load directly from registry.	`None`
`trial`	`Any \| None`	Optional Optuna trial for hyperparameter optimization and pruning.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dictionary with all results (metrics, scores, heatmap, optional anomaly heatmaps).

Source code in app/pipelines/modelling/keras_cae/cae_pipeline.py

def run_keras_cae_pipeline(
    data_root: str = "data/raw/mvtec_ad",
    category: str = "bottle",
    img_size: int = 256,
    crop_size: int = 64,
    crop_stride: int = 32,
    latent_channels: int = 32,
    epochs: int = 20,
    batch_size: int = 16,
    mask_ratio: float = 0.25,
    mask_patch_size: int = 8,
    threshold_method: str = "quantile",
    k_fraction: float = 0.002,
    pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None,
    run_heatmap: bool = False,
    force_retrain: bool = False,
    model_hash: str | None = None,
    trial: Any | None = None,
) -> dict[str, Any]:
    """Run the complete Keras CAE anomaly detection pipeline for one MVTec category.

    This is the main entry point called by the Streamlit application. It:
    1. Resolves cached model or configures new training parameters.
    2. Loads train (normal only) and test images as numpy arrays using exact parameters.
    3. Applies modular preprocessing transforms consistent with model state.
    4. Normalises images to [0, 1].
    5. Builds and trains the Keras CAE with MIM + SSIM+MSE + AdamW (or loads from cache).
    6. Scores all test images using Top-K pooling.
    7. Computes an adaptive threshold from normal validation scores.
    8. Evaluates with image-level AUROC and pixel-level AUPIMO.
    9. Optionally computes Reconstruction Error Heatmap overlays for every anomalous test image.

    Args:
        data_root: Path to the MVTec AD dataset root directory.
        category: MVTec category to train and evaluate on (e.g., 'bottle', 'wood').
        img_size: Size (height and width) to resize base images to.
        crop_size: Size of the sliding window crops extracted from the base image.
        crop_stride: Stride of the sliding window.
        latent_channels: Number of channels in the convolutional bottleneck.
        epochs: Number of training epochs.
        batch_size: Training batch size (number of crops, not full images).
        mask_ratio: Fraction of patches to mask during Masked Image Modeling training.
        mask_patch_size: Side length of each masked region within a crop.
        threshold_method: ``"quantile"`` or ``"mahalanobis"`` for adaptive threshold.
        k_fraction: Top-K fraction for image-level anomaly score pooling.
        pipeline: Optional configuration list or PreprocessingPipeline object.
        run_heatmap: Whether to compute Reconstruction Error heatmap overlays for anomalous images.
        force_retrain: If True, bypass the cache and force training of a new model.
        model_hash: Optional specific model hash to load directly from registry.
        trial: Optional Optuna trial for hyperparameter optimization and pruning.

    Returns:
        Dictionary with all results (metrics, scores, heatmap, optional anomaly heatmaps).
    """
    tf = _require_tf()

    manifest = build_mvtec_manifest(data_root)
    fair_split = build_fair_evaluation_split(manifest, category)
    split_evidence = fair_split.evidence()
    cache_evidence = {**split_evidence, **fair_metric_evidence()}

    cached, cfg, proc_pipeline, norm_prep = _resolve_cae_cache_and_config(
        category=category,
        img_size=img_size,
        crop_size=crop_size,
        crop_stride=crop_stride,
        latent_channels=latent_channels,
        epochs=epochs,
        batch_size=batch_size,
        mask_ratio=mask_ratio,
        mask_patch_size=mask_patch_size,
        threshold_method=threshold_method,
        k_fraction=k_fraction,
        pipeline=pipeline,
        force_retrain=force_retrain,
        model_hash=model_hash,
        cache_evidence=cache_evidence,
    )

    if cached is not None:
        registry_dir, meta = cached
        logger.info(
            "Found newest cached model matching parameters (Hash: %s, Dir: %s). Loading from disk...",
            cfg["model_hash"],
            registry_dir,
        )
        model = tf.keras.models.load_model(cfg["model_path"], compile=False)
        loss_history = meta.get("loss_history", {"train": [], "val_good": [], "val_anomalous": []})
        active_meta = meta
    else:
        loss_history = {"train": [], "val_good": [], "val_anomalous": []}
        active_meta = {}

    logger.info(
        "=== Keras CAE Pipeline: category='%s', img_size=%d, hash='%s' ===",
        cfg["category"],
        cfg["img_size"],
        cfg["model_hash"],
    )

    train_paths = fair_split.fitting_paths
    val_paths = fair_split.validation_paths
    test_df = fair_split.test
    test_paths = fair_split.test_paths
    test_labels = test_df["is_anomaly"].astype(int).to_numpy()
    mask_paths = test_df["mask_path"].tolist()

    logger.info("Train (normal): %d | Val (normal): %d | Test: %d", len(train_paths), len(val_paths), len(test_paths))

    if len(proc_pipeline) > 0:
        logger.info("Applying %d preprocessing steps.", len(proc_pipeline))

    train_images_uint8 = _load_images_as_numpy(train_paths, cfg["img_size"], proc_pipeline)
    val_images_uint8 = _load_images_as_numpy(val_paths, cfg["img_size"], proc_pipeline)
    test_images_uint8 = _load_images_as_numpy(test_paths, cfg["img_size"], proc_pipeline)

    augmenter = get_augmenter(cfg["category"])
    augmented = augment_batch(train_images_uint8, augmenter)
    train_images_uint8 = np.concatenate([train_images_uint8, augmented], axis=0)
    logger.info("After augmentation: %d training images.", len(train_images_uint8))

    train_images = train_images_uint8.astype(np.float32) / 255.0
    test_images = test_images_uint8.astype(np.float32) / 255.0
    val_good_images = val_images_uint8.astype(np.float32) / 255.0

    train_crops = extract_crops(train_images, cfg["crop_size"], cfg["crop_stride"])
    val_good_crops = (
        extract_crops(val_good_images, cfg["crop_size"], cfg["crop_stride"]) if len(val_good_images) > 0 else None
    )

    dataset_split = {
        **cache_evidence,
        "test_normal": int(sum(1 for label_val in test_labels if label_val == 0)),
        "test_anomalous": int(sum(1 for label_val in test_labels if label_val == 1)),
    }

    if cached is None:
        model, loss_history, active_meta = _train_and_save_cae_model(
            cfg=cfg,
            norm_prep=norm_prep,
            dataset_split=dataset_split,
            train_crops=train_crops,
            val_good_crops=val_good_crops,
            val_an_crops=None,
            trial=trial,
        )

    logger.info("Extracting crops for val_good images and predicting...")
    val_good_reconstructed_crops = model.predict(val_good_crops, batch_size=cfg["batch_size"], verbose=0)
    val_good_reconstructed = stitch_crops(
        val_good_reconstructed_crops,
        len(val_good_images),
        cfg["img_size"],
        cfg["img_size"],
        cfg["crop_size"],
        cfg["crop_stride"],
    )

    normal_scores, _ = compute_image_scores(
        model, val_good_images, k_fraction=cfg["k_fraction"], reconstructions=val_good_reconstructed
    )
    threshold = compute_adaptive_threshold(normal_scores, method=cfg["threshold_method"])

    logger.info("Extracting crops for all test images and predicting...")
    test_crops = extract_crops(test_images, cfg["crop_size"], cfg["crop_stride"])
    test_reconstructed_crops = model.predict(test_crops, batch_size=cfg["batch_size"], verbose=0)
    test_reconstructed = stitch_crops(
        test_reconstructed_crops,
        len(test_images),
        cfg["img_size"],
        cfg["img_size"],
        cfg["crop_size"],
        cfg["crop_stride"],
    )

    gt_masks = _load_masks_as_numpy(mask_paths, cfg["img_size"])
    results = evaluate_cae(
        model=model,
        test_images=test_images,
        test_labels=test_labels,
        gt_masks=gt_masks,
        threshold=threshold,
        k_fraction=cfg["k_fraction"],
        output_dir=cfg["registry_dir"],
        reconstructions=test_reconstructed,
    )

    results = _build_cae_result_dict(
        results=results,
        cfg=cfg,
        loss_history=loss_history,
        active_meta=active_meta,
        dataset_split=dataset_split,
        threshold=threshold,
        norm_prep=norm_prep,
        test_images=test_images,
        test_labels=test_labels,
    )

    if run_heatmap and results["anomalous_indices"]:
        results["heatmap_overlays"] = _compute_cae_heatmaps(
            model=model,
            test_images=test_images,
            test_reconstructed=test_reconstructed,
            gt_masks=gt_masks,
            anomalous_indices=results["anomalous_indices"],
        )

    return results

`app.pipelines.modelling.keras_cae.optuna_study` ¶

`objective(trial: optuna.Trial, category_name: str, data_root: str = 'data/raw/mvtec_ad') -> float` ¶

Optuna objective function for optimizing Keras CAE hyperparameters.

Parameters:

Name	Type	Description	Default
`trial`	`Trial`	Optuna trial object.	required
`category_name`	`str`	MVTec category name to optimize.	required
`data_root`	`str`	Path to the MVTec AD dataset.	`'data/raw/mvtec_ad'`

Returns:

Type	Description
`float`	Pixel AUPIMO score to maximize.

Source code in app/pipelines/modelling/keras_cae/optuna_study.py

def objective(trial: optuna.Trial, category_name: str, data_root: str = "data/raw/mvtec_ad") -> float:
    """Optuna objective function for optimizing Keras CAE hyperparameters.

    Args:
        trial: Optuna trial object.
        category_name: MVTec category name to optimize.
        data_root: Path to the MVTec AD dataset.

    Returns:
        Pixel AUPIMO score to maximize.
    """
    # 1. Hyperparameter Search Space
    latent_channels = trial.suggest_categorical("latent_channels", [16, 32, 64, 128])
    apply_clahe = trial.suggest_categorical("apply_clahe", [True, False])
    apply_blur = trial.suggest_categorical("apply_blur", [True, False])

    blur_ksize = 5
    if apply_blur:
        blur_ksize = trial.suggest_categorical("blur_ksize", [3, 5, 7])

    # Type-Specific Logic for MVTec categories
    if category_name in TEXTURE_CATEGORIES:
        apply_foreground_mask = False
    else:
        apply_foreground_mask = trial.suggest_categorical("apply_foreground_mask", [True, False])

    # Build preprocessing steps
    pipeline = []
    if apply_foreground_mask:
        pipeline.append({"name": "foreground_mask", "params": {}})
    if apply_clahe:
        pipeline.append({"name": "clahe", "params": {}})
    if apply_blur:
        pipeline.append({"name": "gaussian_blur", "params": {"kernel_size": blur_ksize}})

    # We use fewer epochs and a smaller batch size to quickly prune bad trials
    epochs = 20
    batch_size = 16

    try:
        results = run_keras_cae_pipeline(
            data_root=data_root,
            category=category_name,
            latent_channels=latent_channels,
            pipeline=pipeline,
            epochs=epochs,
            batch_size=batch_size,
            force_retrain=True,  # Force retrain so it explores the space
            trial=trial,  # Pass trial down to trigger native pruning
        )
    except optuna.exceptions.TrialPruned:
        raise
    except Exception as e:
        logger.error("Trial failed during execution: %s", e)
        raise optuna.exceptions.TrialPruned() from e
    finally:
        import gc

        from tensorflow.keras import backend

        backend.clear_session()
        gc.collect()

    # Extract the target metric to maximize
    pixel_aupimo = results.get("pixel_level", {}).get("aupimo")

    if pixel_aupimo is None:
        raise ValueError("Pixel AUPIMO score not found in results.")

    return float(pixel_aupimo)

`run_study(category_name: str, n_trials: int = 15, data_root: str = 'data/raw/mvtec_ad') -> dict[str, Any]` ¶

Run the Optuna study and save the best parameters.

This function uses Optuna to find the best hyperparameters for the Keras CAE model for a specific MVTec AD category. It uses the fair-eval-v1 protocol to evaluate the model and prunes trials that are unlikely to yield good results.

Parameters:

Name	Type	Description	Default
`category_name`	`str`	MVTec category name to optimize.	required
`n_trials`	`int`	Number of trials to run (default: 15).	`15`
`data_root`	`str`	Path to the MVTec AD dataset.	`'data/raw/mvtec_ad'`

Returns:

Type	Description
`dict[str, Any]`	Best parameters dictionary.

Source code in app/pipelines/modelling/keras_cae/optuna_study.py

def run_study(category_name: str, n_trials: int = 15, data_root: str = "data/raw/mvtec_ad") -> dict[str, Any]:
    """Run the Optuna study and save the best parameters.

    This function uses Optuna to find the best hyperparameters for the Keras CAE model
    for a specific MVTec AD category. It uses the fair-eval-v1 protocol to evaluate the
    model and prunes trials that are unlikely to yield good results.

    Args:
        category_name: MVTec category name to optimize.
        n_trials: Number of trials to run (default: 15).
        data_root: Path to the MVTec AD dataset.

    Returns:
        Best parameters dictionary.
    """
    study_name = f"keras_cae_{category_name}"

    storage_path = Path("data/hyperparameters/keras_cae_optuna.db")
    storage_path.parent.mkdir(parents=True, exist_ok=True)
    storage_url = f"sqlite:///{storage_path.resolve()}"

    # We want to MAXIMIZE Pixel AUPIMO
    study = optuna.create_study(
        study_name=study_name,
        storage=storage_url,
        load_if_exists=True,
        direction="maximize",
        pruner=optuna.pruners.MedianPruner(n_startup_trials=3, n_warmup_steps=5, interval_steps=1),
    )

    trials_to_run = max(0, n_trials - len(study.trials))
    if trials_to_run > 0:
        study.optimize(lambda trial: objective(trial, category_name, data_root), n_trials=trials_to_run)

    best_params = study.best_params

    # Map to nested schema
    is_texture = category_name in TEXTURE_CATEGORIES
    use_mask = False if is_texture else best_params.get("apply_foreground_mask", False)

    cfg = {
        "target_metric": "pixel_aupimo",
        "score": study.best_value,
        "preprocessing": {
            "use_foreground_mask": use_mask,
            "use_clahe": best_params.get("apply_clahe", False),
            "clahe_clip_limit": 2.0,
            "clahe_tile_grid_size": [8, 8],
            "use_gaussian_blur": best_params.get("apply_blur", False),
            "blur_ksize": best_params.get("blur_ksize", 5),
        },
        "model_hyperparameters": {
            "learning_rate": 0.001,
            "latent_dim": best_params.get("latent_channels", 32),
            "loss_weight_ssim": 0.84,
            "loss_weight_mse": 0.16,
        },
    }

    # Save to JSON registry
    registry_path = Path("data/hyperparameters/keras_cae_best.json")
    registry_path.parent.mkdir(parents=True, exist_ok=True)

    if registry_path.exists():
        with open(registry_path, encoding="utf-8") as f:
            registry = json.load(f)
    else:
        registry = {}

    registry[category_name] = cfg

    with open(registry_path, "w", encoding="utf-8") as f:
        json.dump(registry, f, indent=4)

    logger.info("Study finished. Best Pixel AUPIMO: %.4f", study.best_value)
    logger.info("Best Params: %s", cfg)

    return cfg

`app.pipelines.modelling.dino.v2` ¶

Frozen DINOv2 patch-token nearest-neighbour baseline for MVTec AD.

run_dinov2_baseline(data_root: Path | str = 'data/raw/mvtec_ad', category: str = 'bottle', pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None, fpr_limit: float = 0.0001, encoder_name: str = DINO_V2_ENCODER, num_neighbors: int = 1, masking: MaskingMode = 'published', run_heatmap: bool = False, preprocessing_steps: list[dict[str, Any]] | None = None, registry_base: Path | str = 'data/models/dinov2', model_seed: int = PATCHCORE_MODEL_SEED, variant: DINOVariant = 'baseline', feature_layers: tuple[int, ...] = ENHANCED_DINO_LAYERS, position_radius: int = 1, spatial_weight: float = 0.05, density_neighbors: int = 5, reuse_complete: bool = False) -> BaselineResult | AllCategoriesResult ¶

Run one MVTec category or all canonical categories with frozen DINOv2.

category="all" orchestrates sequential evaluation across all 15 categories, ensuring each receives its fixed split, normal feature bank, and threshold calibration.

Parameters:

Name	Type	Description	Default
`data_root`	`Path \| str`	Root directory of MVTec AD dataset.	`'data/raw/mvtec_ad'`
`category`	`str`	Category name or 'all' for full benchmark evaluation.	`'bottle'`
`pipeline`	`list[dict[str, Any]] \| PreprocessingPipeline \| None`	Preprocessing pipeline or list of step configs.	`None`
`fpr_limit`	`float`	Fair-eval AUPIMO upper FPR bound (fixed at 1e-4).	`0.0001`
`encoder_name`	`str`	Pretrained DINOv2 vision transformer encoder identifier.	`DINO_V2_ENCODER`
`num_neighbors`	`int`	Number of normal patch neighbours to query.	`1`
`masking`	`MaskingMode`	Foreground PCA masking policy ('off', 'on', or 'published').	`'published'`
`run_heatmap`	`bool`	Whether to render anomalous test-image heatmaps.	`False`
`preprocessing_steps`	`list[dict[str, Any]] \| None`	Legacy parameter preserved for backward compatibility.	`None`
`registry_base`	`Path \| str`	Output root directory for evaluation artifacts.	`'data/models/dinov2'`
`model_seed`	`int`	Deterministic model and data loader seed.	`PATCHCORE_MODEL_SEED`
`variant`	`DINOVariant`	Feature extractor variant ('baseline' or 'enhanced').	`'baseline'`
`feature_layers`	`tuple[int, ...]`	Transformer block indices for enhanced scoring.	`ENHANCED_DINO_LAYERS`
`position_radius`	`int`	Patch search radius for enhanced spatial matching.	`1`
`spatial_weight`	`float`	Penalty weight for spatial distance in enhanced scoring.	`0.05`
`density_neighbors`	`int`	Number of neighbours for local density estimation.	`5`
`reuse_complete`	`bool`	Return saved evaluation artifacts for an exact cache hit.	`False`

Returns:

Type	Description
`BaselineResult \| AllCategoriesResult`	BaselineResult for a single category, or AllCategoriesResult for 'all'.

Source code in app/pipelines/modelling/dino/v2.py

def run_dinov2_baseline(
    data_root: Path | str = "data/raw/mvtec_ad",
    category: str = "bottle",
    pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None,
    fpr_limit: float = 1e-4,
    encoder_name: str = DINO_V2_ENCODER,
    num_neighbors: int = 1,
    masking: MaskingMode = "published",
    run_heatmap: bool = False,
    preprocessing_steps: list[dict[str, Any]] | None = None,
    registry_base: Path | str = "data/models/dinov2",
    model_seed: int = PATCHCORE_MODEL_SEED,
    variant: DINOVariant = "baseline",
    feature_layers: tuple[int, ...] = ENHANCED_DINO_LAYERS,
    position_radius: int = 1,
    spatial_weight: float = 0.05,
    density_neighbors: int = 5,
    reuse_complete: bool = False,
) -> BaselineResult | AllCategoriesResult:
    """Run one MVTec category or all canonical categories with frozen DINOv2.

    ``category="all"`` orchestrates sequential evaluation across all 15 categories,
    ensuring each receives its fixed split, normal feature bank, and threshold calibration.

    Args:
        data_root: Root directory of MVTec AD dataset.
        category: Category name or 'all' for full benchmark evaluation.
        pipeline: Preprocessing pipeline or list of step configs.
        fpr_limit: Fair-eval AUPIMO upper FPR bound (fixed at 1e-4).
        encoder_name: Pretrained DINOv2 vision transformer encoder identifier.
        num_neighbors: Number of normal patch neighbours to query.
        masking: Foreground PCA masking policy ('off', 'on', or 'published').
        run_heatmap: Whether to render anomalous test-image heatmaps.
        preprocessing_steps: Legacy parameter preserved for backward compatibility.
        registry_base: Output root directory for evaluation artifacts.
        model_seed: Deterministic model and data loader seed.
        variant: Feature extractor variant ('baseline' or 'enhanced').
        feature_layers: Transformer block indices for enhanced scoring.
        position_radius: Patch search radius for enhanced spatial matching.
        spatial_weight: Penalty weight for spatial distance in enhanced scoring.
        density_neighbors: Number of neighbours for local density estimation.
        reuse_complete: Return saved evaluation artifacts for an exact cache hit.

    Returns:
        BaselineResult for a single category, or AllCategoriesResult for 'all'.
    """
    if category != "all":
        return run_dino_category(
            data_root=data_root,
            category=category,
            pipeline=pipeline,
            fpr_limit=fpr_limit,
            encoder_name=encoder_name,
            num_neighbors=num_neighbors,
            masking=masking,
            run_heatmap=run_heatmap,
            preprocessing_steps=preprocessing_steps,
            registry_base=registry_base,
            model_seed=model_seed,
            variant=variant,
            feature_layers=feature_layers,
            position_radius=position_radius,
            spatial_weight=spatial_weight,
            density_neighbors=density_neighbors,
            reuse_complete=reuse_complete,
            model_generation="dinov2",
            model_name="DINOv2",
            input_size=DINO_V2_INPUT_SIZE,
            patch_size=DINO_V2_PATCH_SIZE,
            batch_size=DINO_V2_BATCH_SIZE,
        )

    return run_dino_all_categories(
        data_root=data_root,
        pipeline=pipeline,
        fpr_limit=fpr_limit,
        encoder_name=encoder_name,
        num_neighbors=num_neighbors,
        masking=masking,
        run_heatmap=run_heatmap,
        registry_base=registry_base,
        model_seed=model_seed,
        variant=variant,
        feature_layers=feature_layers,
        position_radius=position_radius,
        spatial_weight=spatial_weight,
        density_neighbors=density_neighbors,
        model_generation="dinov2",
        model_name="DINOv2",
        input_size=DINO_V2_INPUT_SIZE,
        patch_size=DINO_V2_PATCH_SIZE,
        batch_size=DINO_V2_BATCH_SIZE,
        reuse_complete=reuse_complete,
        save_summary_files=False,
    )

`app.pipelines.modelling.dino.v3` ¶

Frozen DINOv3 patch-token nearest-neighbour baseline for MVTec AD.

`run_dinov3_baseline(data_root: Path | str = 'data/raw/mvtec_ad', category: str = 'bottle', pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None, fpr_limit: float = 0.0001, encoder_name: str = DINO_V3_ENCODER, num_neighbors: int = 1, masking: MaskingMode = 'off', run_heatmap: bool = False, registry_base: Path | str = 'data/models/dinov3', model_seed: int = PATCHCORE_MODEL_SEED, reuse_complete: bool = False) -> BaselineResult | AllCategoriesResult` ¶

Run the fair frozen-DINOv3 baseline for one or all MVTec categories.

Parameters:

Name	Type	Description	Default
`data_root`	`Path \| str`	Root directory of MVTec AD dataset.	`'data/raw/mvtec_ad'`
`category`	`str`	MVTec category or 'all' for full benchmark evaluation.	`'bottle'`
`pipeline`	`list[dict[str, Any]] \| PreprocessingPipeline \| None`	Optional preprocessing pipeline or configuration list.	`None`
`fpr_limit`	`float`	Fair-eval AUPIMO upper FPR bound (fixed at 1e-4).	`0.0001`
`encoder_name`	`str`	Pretrained DINOv3 encoder identifier.	`DINO_V3_ENCODER`
`num_neighbors`	`int`	Number of normal patch neighbours to query.	`1`
`masking`	`MaskingMode`	Foreground masking policy (must be 'off' for DINOv3).	`'off'`
`run_heatmap`	`bool`	Whether to render test-image heatmaps.	`False`
`registry_base`	`Path \| str`	Target artifact directory.	`'data/models/dinov3'`
`model_seed`	`int`	Deterministic model and data loader seed.	`PATCHCORE_MODEL_SEED`
`reuse_complete`	`bool`	Return saved evaluation artifacts for an exact cache hit.	`False`

Returns:

Type	Description
`BaselineResult \| AllCategoriesResult`	BaselineResult for a single category, or AllCategoriesResult for 'all'.

Source code in app/pipelines/modelling/dino/v3.py

def run_dinov3_baseline(
    data_root: Path | str = "data/raw/mvtec_ad",
    category: str = "bottle",
    pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None,
    fpr_limit: float = 1e-4,
    encoder_name: str = DINO_V3_ENCODER,
    num_neighbors: int = 1,
    masking: MaskingMode = "off",
    run_heatmap: bool = False,
    registry_base: Path | str = "data/models/dinov3",
    model_seed: int = PATCHCORE_MODEL_SEED,
    reuse_complete: bool = False,
) -> BaselineResult | AllCategoriesResult:
    """Run the fair frozen-DINOv3 baseline for one or all MVTec categories.

    Args:
        data_root: Root directory of MVTec AD dataset.
        category: MVTec category or 'all' for full benchmark evaluation.
        pipeline: Optional preprocessing pipeline or configuration list.
        fpr_limit: Fair-eval AUPIMO upper FPR bound (fixed at 1e-4).
        encoder_name: Pretrained DINOv3 encoder identifier.
        num_neighbors: Number of normal patch neighbours to query.
        masking: Foreground masking policy (must be 'off' for DINOv3).
        run_heatmap: Whether to render test-image heatmaps.
        registry_base: Target artifact directory.
        model_seed: Deterministic model and data loader seed.
        reuse_complete: Return saved evaluation artifacts for an exact cache hit.

    Returns:
        BaselineResult for a single category, or AllCategoriesResult for 'all'.
    """
    if category != "all":
        return _run_dinov3_category(
            data_root=data_root,
            category=category,
            pipeline=pipeline,
            fpr_limit=fpr_limit,
            encoder_name=encoder_name,
            num_neighbors=num_neighbors,
            masking=masking,
            run_heatmap=run_heatmap,
            registry_base=registry_base,
            model_seed=model_seed,
            reuse_complete=reuse_complete,
        )

    if masking != "off":
        raise ValueError(
            "DINOv3 requires masking='off': Anomalib's published PCA threshold is calibrated for DINOv2 "
            "and can produce an empty DINOv3 memory bank."
        )

    return run_dino_all_categories(
        data_root=data_root,
        pipeline=pipeline,
        fpr_limit=fpr_limit,
        encoder_name=encoder_name,
        num_neighbors=num_neighbors,
        masking=masking,
        run_heatmap=run_heatmap,
        registry_base=registry_base,
        model_seed=model_seed,
        variant="baseline",
        feature_layers=(),
        position_radius=0,
        spatial_weight=0.0,
        density_neighbors=0,
        model_generation="dinov3",
        model_name="DINOv3",
        input_size=DINO_V3_INPUT_SIZE,
        patch_size=DINO_V3_PATCH_SIZE,
        batch_size=4,
        reuse_complete=reuse_complete,
        save_summary_files=True,
    )

`app.pipelines.modelling.dino.enhanced` ¶

Multi-layer, position-aware DINOv2 anomaly scoring.

`EnhancedAnomalyDINOModel` ¶

Bases: AnomalyDINOModel

DINOv2 patch bank with multi-layer and spatially local density-aware kNN.

Source code in app/pipelines/modelling/dino/enhanced.py

class EnhancedAnomalyDINOModel(AnomalyDINOModel):
    """DINOv2 patch bank with multi-layer and spatially local density-aware kNN."""

    def __init__(
        self,
        num_neighbours: int = 5,
        encoder_name: str = "vit_small_patch14_dinov2",
        masking: bool = False,
        feature_layers: Sequence[int] = (8, 10, 11),
        position_radius: int = 1,
        spatial_weight: float = 0.05,
        density_neighbours: int = 5,
    ) -> None:
        """Configure the frozen layers and spatial-density kNN scorer.

        Args:
            num_neighbours: Neighbours averaged for each query patch.
            encoder_name: Pretrained timm DINOv2 encoder name.
            masking: Whether to apply AnomalyDINO's PCA foreground mask.
            feature_layers: Zero-based transformer block indices to concatenate.
            position_radius: Maximum row/column offset for candidate patches.
            spatial_weight: Additive penalty per squared patch-grid offset.
            density_neighbours: Neighbours used for normal-density estimation.
        """
        super().__init__(
            num_neighbours=num_neighbours,
            encoder_name=encoder_name,
            masking=masking,
            coreset_subsampling=False,
        )
        if not feature_layers:
            raise ValueError("feature_layers must not be empty")
        if position_radius < 0:
            raise ValueError("position_radius must be non-negative")
        if spatial_weight < 0:
            raise ValueError("spatial_weight must be non-negative")
        if density_neighbours < 1:
            raise ValueError("density_neighbours must be at least 1")

        self.feature_layers = tuple(int(layer) for layer in feature_layers)
        self.layer_names = tuple(f"blocks.{layer}" for layer in self.feature_layers)
        self.position_radius = position_radius
        self.spatial_weight = spatial_weight
        self.density_neighbours = density_neighbours
        self.feature_encoder = TimmFeatureExtractor(
            backbone=encoder_name,
            layers=list(self.layer_names),
            pre_trained=True,
            requires_grad=False,
            output_fmt="NLC",
            return_class_token=False,
            norm=True,
            dynamic_img_size=True,
        )
        self.patch_size = self.feature_encoder.patch_size
        self.embedding_store: list[torch.Tensor] = []
        self.mask_store: list[torch.Tensor] = []
        self.register_buffer("memory_valid", torch.empty(0, dtype=torch.bool))
        self.register_buffer("memory_density", torch.empty(0))
        self.register_buffer("density_reference", torch.tensor(1.0))

    def extract_features(self, image_tensor: torch.Tensor) -> torch.Tensor:
        """Concatenate raw patch tokens from the selected transformer blocks."""
        outputs = self.feature_encoder(image_tensor)
        return torch.cat([outputs[name] for name in self.layer_names], dim=-1)

    def fit(self) -> None:
        """Finalize the structured bank and estimate normal density at each position."""
        if not self.embedding_store:
            raise ValueError("No embeddings collected. Run model in training mode first.")
        self.memory_bank = torch.cat(self.embedding_store, dim=0)
        self.memory_valid = torch.cat(self.mask_store, dim=0)
        self.embedding_store.clear()
        self.mask_store.clear()

        samples, patches, _ = self.memory_bank.shape
        densities = torch.ones((samples, patches), device=self.memory_bank.device, dtype=self.memory_bank.dtype)
        for patch_index in range(patches):
            valid_indices = torch.nonzero(self.memory_valid[:, patch_index], as_tuple=False).squeeze(1)
            if valid_indices.numel() < 2:
                continue
            features = self.memory_bank[valid_indices, patch_index]
            distances = (1 - features @ features.T).clamp_(0, 2)
            distances.fill_diagonal_(float("inf"))
            k = min(self.density_neighbours, valid_indices.numel() - 1)
            local_density = distances.topk(k=k, largest=False, dim=1).values.mean(dim=1)
            densities[valid_indices, patch_index] = local_density.clamp_min(1e-3)
        self.memory_density = densities
        valid_densities = densities[self.memory_valid]
        self.density_reference = valid_densities.median().clamp_min(1e-3)

    @staticmethod
    def patchcore_image_score(neighbour_scores: torch.Tensor) -> torch.Tensor:
        """Apply PatchCore-style neighborhood confidence to the worst patch."""
        patch_scores = neighbour_scores.mean(dim=-1)
        if neighbour_scores.shape[-1] == 1:
            return patch_scores.amax(dim=1, keepdim=True)
        worst_patch = patch_scores.argmax(dim=1)
        batch_indices = torch.arange(patch_scores.shape[0], device=patch_scores.device)
        worst_neighbours = neighbour_scores[batch_indices, worst_patch]
        confidence = 1 - F.softmax(worst_neighbours, dim=1)[:, 0]
        image_score: torch.Tensor = (confidence * patch_scores[batch_indices, worst_patch]).unsqueeze(1)
        return image_score

    def _score_features(
        self,
        features: torch.Tensor,
        grid_size: tuple[int, int],
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Return density-normalized kNN scores for every spatial patch."""
        batch_size, num_patches, _ = features.shape
        height, width = grid_size
        k = min(self.num_neighbours, self.memory_bank.shape[0] * (2 * self.position_radius + 1) ** 2)
        neighbour_scores = torch.zeros((batch_size, num_patches, k), device=features.device, dtype=features.dtype)

        for patch_index in range(num_patches):
            row, column = divmod(patch_index, width)
            nearby = [
                y * width + x
                for y in range(max(0, row - self.position_radius), min(height, row + self.position_radius + 1))
                for x in range(max(0, column - self.position_radius), min(width, column + self.position_radius + 1))
            ]
            candidates = self.memory_bank[:, nearby].reshape(-1, self.memory_bank.shape[-1])
            candidate_valid = self.memory_valid[:, nearby].reshape(-1)
            candidate_density = self.memory_density[:, nearby].reshape(-1)
            candidate_positions = torch.tensor(nearby, device=features.device).repeat(self.memory_bank.shape[0])
            candidates = candidates[candidate_valid]
            candidate_density = candidate_density[candidate_valid]
            candidate_positions = candidate_positions[candidate_valid]
            if candidates.shape[0] < k:
                candidates = self.memory_bank.reshape(-1, self.memory_bank.shape[-1])
                candidate_valid = self.memory_valid.reshape(-1)
                candidate_density = self.memory_density.reshape(-1)
                candidate_positions = torch.arange(num_patches, device=features.device).repeat(
                    self.memory_bank.shape[0]
                )
                candidates = candidates[candidate_valid]
                candidate_density = candidate_density[candidate_valid]
                candidate_positions = candidate_positions[candidate_valid]
            if candidates.shape[0] < k:
                raise RuntimeError("Too few valid DINOv2 neighbours after global fallback")

            distances = (1 - features[:, patch_index] @ candidates.T).clamp_(0, 2)
            candidate_rows = torch.div(candidate_positions, width, rounding_mode="floor")
            candidate_cols = candidate_positions.remainder(width)
            spatial_distance = (candidate_rows - row).square() + (candidate_cols - column).square()
            distances = distances + self.spatial_weight * spatial_distance.to(distances.dtype)
            values, indices = distances.topk(k=k, largest=False, dim=1)
            density_scale = (self.density_reference / candidate_density[indices].clamp_min(1e-3)).pow(0.25)
            neighbour_scores[:, patch_index] = values * density_scale.clamp(0.5, 2.0)

        return neighbour_scores.mean(dim=-1), neighbour_scores

    def forward(self, input_tensor: torch.Tensor) -> torch.Tensor | InferenceBatch:
        """Collect structured normal tokens or score test tokens."""
        input_tensor = input_tensor.type(self.memory_bank.dtype)
        batch_size, _, input_height, input_width = input_tensor.shape
        crop_height = input_height % self.patch_size
        crop_width = input_width % self.patch_size
        top, left = crop_height // 2, crop_width // 2
        bottom, right = crop_height - top, crop_width - left
        cropped_height, cropped_width = input_height - crop_height, input_width - crop_width
        if crop_height or crop_width:
            input_tensor = input_tensor[:, :, top : input_height - bottom, left : input_width - right]
        grid_size = (cropped_height // self.patch_size, cropped_width // self.patch_size)

        features = self.extract_features(input_tensor)
        if self.masking:
            masks_np = self.compute_background_masks(features.detach().cpu().numpy(), grid_size)
            masks = torch.from_numpy(masks_np).to(features.device)
        else:
            masks = torch.ones(features.shape[:2], dtype=torch.bool, device=features.device)
        features = F.normalize(features, p=2, dim=-1)

        if self.training:
            self.embedding_store.append(features)
            self.mask_store.append(masks)
            return torch.tensor(0.0, device=features.device, requires_grad=True)
        if self.memory_bank.numel() == 0:
            raise RuntimeError("Memory bank is empty. Run fit before inference.")

        patch_scores, neighbour_scores = self._score_features(features, grid_size)
        patch_scores = patch_scores.masked_fill(~masks, 0)
        neighbour_scores = neighbour_scores.masked_fill(~masks.unsqueeze(-1), 0)
        image_score = self.patchcore_image_score(neighbour_scores)
        anomaly_map = patch_scores.view(batch_size, 1, *grid_size)
        anomaly_map = self.anomaly_map_generator(anomaly_map, (cropped_height, cropped_width))
        if crop_height or crop_width:
            anomaly_map = F.pad(anomaly_map, (left, right, top, bottom), mode="replicate")
        return InferenceBatch(pred_score=image_score, anomaly_map=anomaly_map)

`init(num_neighbours: int = 5, encoder_name: str = 'vit_small_patch14_dinov2', masking: bool = False, feature_layers: Sequence[int] = (8, 10, 11), position_radius: int = 1, spatial_weight: float = 0.05, density_neighbours: int = 5) -> None` ¶

Configure the frozen layers and spatial-density kNN scorer.

Parameters:

Name	Type	Description	Default
`num_neighbours`	`int`	Neighbours averaged for each query patch.	`5`
`encoder_name`	`str`	Pretrained timm DINOv2 encoder name.	`'vit_small_patch14_dinov2'`
`masking`	`bool`	Whether to apply AnomalyDINO's PCA foreground mask.	`False`
`feature_layers`	`Sequence[int]`	Zero-based transformer block indices to concatenate.	`(8, 10, 11)`
`position_radius`	`int`	Maximum row/column offset for candidate patches.	`1`
`spatial_weight`	`float`	Additive penalty per squared patch-grid offset.	`0.05`
`density_neighbours`	`int`	Neighbours used for normal-density estimation.	`5`

Source code in app/pipelines/modelling/dino/enhanced.py

def __init__(
    self,
    num_neighbours: int = 5,
    encoder_name: str = "vit_small_patch14_dinov2",
    masking: bool = False,
    feature_layers: Sequence[int] = (8, 10, 11),
    position_radius: int = 1,
    spatial_weight: float = 0.05,
    density_neighbours: int = 5,
) -> None:
    """Configure the frozen layers and spatial-density kNN scorer.

    Args:
        num_neighbours: Neighbours averaged for each query patch.
        encoder_name: Pretrained timm DINOv2 encoder name.
        masking: Whether to apply AnomalyDINO's PCA foreground mask.
        feature_layers: Zero-based transformer block indices to concatenate.
        position_radius: Maximum row/column offset for candidate patches.
        spatial_weight: Additive penalty per squared patch-grid offset.
        density_neighbours: Neighbours used for normal-density estimation.
    """
    super().__init__(
        num_neighbours=num_neighbours,
        encoder_name=encoder_name,
        masking=masking,
        coreset_subsampling=False,
    )
    if not feature_layers:
        raise ValueError("feature_layers must not be empty")
    if position_radius < 0:
        raise ValueError("position_radius must be non-negative")
    if spatial_weight < 0:
        raise ValueError("spatial_weight must be non-negative")
    if density_neighbours < 1:
        raise ValueError("density_neighbours must be at least 1")

    self.feature_layers = tuple(int(layer) for layer in feature_layers)
    self.layer_names = tuple(f"blocks.{layer}" for layer in self.feature_layers)
    self.position_radius = position_radius
    self.spatial_weight = spatial_weight
    self.density_neighbours = density_neighbours
    self.feature_encoder = TimmFeatureExtractor(
        backbone=encoder_name,
        layers=list(self.layer_names),
        pre_trained=True,
        requires_grad=False,
        output_fmt="NLC",
        return_class_token=False,
        norm=True,
        dynamic_img_size=True,
    )
    self.patch_size = self.feature_encoder.patch_size
    self.embedding_store: list[torch.Tensor] = []
    self.mask_store: list[torch.Tensor] = []
    self.register_buffer("memory_valid", torch.empty(0, dtype=torch.bool))
    self.register_buffer("memory_density", torch.empty(0))
    self.register_buffer("density_reference", torch.tensor(1.0))

`extract_features(image_tensor: torch.Tensor) -> torch.Tensor` ¶

Concatenate raw patch tokens from the selected transformer blocks.

Source code in app/pipelines/modelling/dino/enhanced.py

def extract_features(self, image_tensor: torch.Tensor) -> torch.Tensor:
    """Concatenate raw patch tokens from the selected transformer blocks."""
    outputs = self.feature_encoder(image_tensor)
    return torch.cat([outputs[name] for name in self.layer_names], dim=-1)

`fit() -> None` ¶

Finalize the structured bank and estimate normal density at each position.

Source code in app/pipelines/modelling/dino/enhanced.py

def fit(self) -> None:
    """Finalize the structured bank and estimate normal density at each position."""
    if not self.embedding_store:
        raise ValueError("No embeddings collected. Run model in training mode first.")
    self.memory_bank = torch.cat(self.embedding_store, dim=0)
    self.memory_valid = torch.cat(self.mask_store, dim=0)
    self.embedding_store.clear()
    self.mask_store.clear()

    samples, patches, _ = self.memory_bank.shape
    densities = torch.ones((samples, patches), device=self.memory_bank.device, dtype=self.memory_bank.dtype)
    for patch_index in range(patches):
        valid_indices = torch.nonzero(self.memory_valid[:, patch_index], as_tuple=False).squeeze(1)
        if valid_indices.numel() < 2:
            continue
        features = self.memory_bank[valid_indices, patch_index]
        distances = (1 - features @ features.T).clamp_(0, 2)
        distances.fill_diagonal_(float("inf"))
        k = min(self.density_neighbours, valid_indices.numel() - 1)
        local_density = distances.topk(k=k, largest=False, dim=1).values.mean(dim=1)
        densities[valid_indices, patch_index] = local_density.clamp_min(1e-3)
    self.memory_density = densities
    valid_densities = densities[self.memory_valid]
    self.density_reference = valid_densities.median().clamp_min(1e-3)

`forward(input_tensor: torch.Tensor) -> torch.Tensor | InferenceBatch` ¶

Collect structured normal tokens or score test tokens.

Source code in app/pipelines/modelling/dino/enhanced.py

def forward(self, input_tensor: torch.Tensor) -> torch.Tensor | InferenceBatch:
    """Collect structured normal tokens or score test tokens."""
    input_tensor = input_tensor.type(self.memory_bank.dtype)
    batch_size, _, input_height, input_width = input_tensor.shape
    crop_height = input_height % self.patch_size
    crop_width = input_width % self.patch_size
    top, left = crop_height // 2, crop_width // 2
    bottom, right = crop_height - top, crop_width - left
    cropped_height, cropped_width = input_height - crop_height, input_width - crop_width
    if crop_height or crop_width:
        input_tensor = input_tensor[:, :, top : input_height - bottom, left : input_width - right]
    grid_size = (cropped_height // self.patch_size, cropped_width // self.patch_size)

    features = self.extract_features(input_tensor)
    if self.masking:
        masks_np = self.compute_background_masks(features.detach().cpu().numpy(), grid_size)
        masks = torch.from_numpy(masks_np).to(features.device)
    else:
        masks = torch.ones(features.shape[:2], dtype=torch.bool, device=features.device)
    features = F.normalize(features, p=2, dim=-1)

    if self.training:
        self.embedding_store.append(features)
        self.mask_store.append(masks)
        return torch.tensor(0.0, device=features.device, requires_grad=True)
    if self.memory_bank.numel() == 0:
        raise RuntimeError("Memory bank is empty. Run fit before inference.")

    patch_scores, neighbour_scores = self._score_features(features, grid_size)
    patch_scores = patch_scores.masked_fill(~masks, 0)
    neighbour_scores = neighbour_scores.masked_fill(~masks.unsqueeze(-1), 0)
    image_score = self.patchcore_image_score(neighbour_scores)
    anomaly_map = patch_scores.view(batch_size, 1, *grid_size)
    anomaly_map = self.anomaly_map_generator(anomaly_map, (cropped_height, cropped_width))
    if crop_height or crop_width:
        anomaly_map = F.pad(anomaly_map, (left, right, top, bottom), mode="replicate")
    return InferenceBatch(pred_score=image_score, anomaly_map=anomaly_map)

`patchcore_image_score(neighbour_scores: torch.Tensor) -> torch.Tensor` `staticmethod` ¶

Apply PatchCore-style neighborhood confidence to the worst patch.

Source code in app/pipelines/modelling/dino/enhanced.py

@staticmethod
def patchcore_image_score(neighbour_scores: torch.Tensor) -> torch.Tensor:
    """Apply PatchCore-style neighborhood confidence to the worst patch."""
    patch_scores = neighbour_scores.mean(dim=-1)
    if neighbour_scores.shape[-1] == 1:
        return patch_scores.amax(dim=1, keepdim=True)
    worst_patch = patch_scores.argmax(dim=1)
    batch_indices = torch.arange(patch_scores.shape[0], device=patch_scores.device)
    worst_neighbours = neighbour_scores[batch_indices, worst_patch]
    confidence = 1 - F.softmax(worst_neighbours, dim=1)[:, 0]
    image_score: torch.Tensor = (confidence * patch_scores[batch_indices, worst_patch]).unsqueeze(1)
    return image_score

`app.pipelines.modelling.dino.engine` ¶

Shared execution and evaluation engine for DINO foundation model baselines.

`build_dino_identity_and_hash(category: str, encoder_name: str, num_neighbors: int, masking: MaskingMode, use_masking: bool, raw_prep_list: list[dict[str, Any]], cache_evidence: dict[str, Any], model_generation: Literal['dinov2', 'dinov3'], variant: DINOVariant, feature_layers: tuple[int, ...], position_radius: int, spatial_weight: float, density_neighbors: int) -> tuple[dict[str, Any], str, dict[str, Any]]` ¶

Build the serializable identity dictionary and unique hash for a DINO experiment.

Parameters:

Name	Type	Description	Default
`category`	`str`	MVTec category being evaluated.	required
`encoder_name`	`str`	Name of the pretrained feature encoder.	required
`num_neighbors`	`int`	Nearest neighbors count.	required
`masking`	`MaskingMode`	Active masking policy mode.	required
`use_masking`	`bool`	Evaluated boolean decision for masking.	required
`raw_prep_list`	`list[dict[str, Any]]`	Normalized preprocessing configurations.	required
`cache_evidence`	`dict[str, Any]`	Protocol split and metric version metadata.	required
`model_generation`	`Literal['dinov2', 'dinov3']`	Architectural generation ('dinov2' or 'dinov3').	required
`variant`	`DINOVariant`	Feature extractor variant ('baseline' or 'enhanced').	required
`feature_layers`	`tuple[int, ...]`	Block indices for enhanced scoring.	required
`position_radius`	`int`	Grid search radius for spatial matching.	required
`spatial_weight`	`float`	Weight factor for spatial distance penalty.	required
`density_neighbors`	`int`	Neighbor count for density estimation.	required

Returns:

Type	Description
`tuple[dict[str, Any], str, dict[str, Any]]`	A tuple of (identity_dictionary, 12_char_hex_hash, enhanced_scorer_parameters).

Raises:

Type	Description
`ValueError`	If variant is invalid.

Source code in app/pipelines/modelling/dino/engine.py

def build_dino_identity_and_hash(
    category: str,
    encoder_name: str,
    num_neighbors: int,
    masking: MaskingMode,
    use_masking: bool,
    raw_prep_list: list[dict[str, Any]],
    cache_evidence: dict[str, Any],
    model_generation: Literal["dinov2", "dinov3"],
    variant: DINOVariant,
    feature_layers: tuple[int, ...],
    position_radius: int,
    spatial_weight: float,
    density_neighbors: int,
) -> tuple[dict[str, Any], str, dict[str, Any]]:
    """Build the serializable identity dictionary and unique hash for a DINO experiment.

    Args:
        category: MVTec category being evaluated.
        encoder_name: Name of the pretrained feature encoder.
        num_neighbors: Nearest neighbors count.
        masking: Active masking policy mode.
        use_masking: Evaluated boolean decision for masking.
        raw_prep_list: Normalized preprocessing configurations.
        cache_evidence: Protocol split and metric version metadata.
        model_generation: Architectural generation ('dinov2' or 'dinov3').
        variant: Feature extractor variant ('baseline' or 'enhanced').
        feature_layers: Block indices for enhanced scoring.
        position_radius: Grid search radius for spatial matching.
        spatial_weight: Weight factor for spatial distance penalty.
        density_neighbors: Neighbor count for density estimation.

    Returns:
        A tuple of (identity_dictionary, 12_char_hex_hash, enhanced_scorer_parameters).

    Raises:
        ValueError: If variant is invalid.
    """
    identity: dict[str, Any] = {
        "category": category,
        "encoder_name": encoder_name,
        "num_neighbors": num_neighbors,
        "masking_mode": masking,
        "masking": use_masking,
        "coreset_subsampling": False,
        "preprocessing_steps": raw_prep_list,
        "evaluation": cache_evidence,
    }
    if model_generation != "dinov2":
        identity["model_generation"] = model_generation

    enhanced_scorer: dict[str, Any] = {}
    if variant == "enhanced":
        enhanced_scorer = {
            "feature_layers": list(feature_layers),
            "feature_normalization": "l2_after_pca_mask",
            "position_radius": position_radius,
            "spatial_weight": spatial_weight,
            "position_fallback": "global_with_spatial_penalty",
            "density_neighbors": density_neighbors,
            "density_normalization": "global_median_quarter_power_clipped_0.5_2.0",
            "image_aggregation": "patchcore_neighborhood_reweighting",
        }
        identity["enhanced_scorer"] = enhanced_scorer
    elif variant != "baseline":
        raise ValueError("variant must be one of: baseline, enhanced")

    model_hash = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest()[:12]
    return identity, model_hash, enhanced_scorer

`compute_macro_average(category_results: dict[str, BaselineResult]) -> dict[str, float]` ¶

Compute unweighted macro averages across all evaluated categories.

Parameters:

Name	Type	Description	Default
`category_results`	`dict[str, BaselineResult]`	Mapping of category names to BaselineResults.	required

Returns:

Type	Description
`dict[str, float]`	Dictionary of mean values for image- and pixel-level metrics.

Source code in app/pipelines/modelling/dino/engine.py

def compute_macro_average(category_results: dict[str, BaselineResult]) -> dict[str, float]:
    """Compute unweighted macro averages across all evaluated categories.

    Args:
        category_results: Mapping of category names to BaselineResults.

    Returns:
        Dictionary of mean values for image- and pixel-level metrics.
    """
    macro_average: dict[str, float] = {}
    for metric, (level, key) in METRIC_PATHS.items():
        values = [float(result[level][key]) for result in category_results.values()]  # type: ignore[literal-required]
        macro_average[metric] = float(np.mean(values))
    return macro_average

`instantiate_dino_model(num_neighbors: int, encoder_name: str, use_masking: bool, model_generation: Literal['dinov2', 'dinov3'], variant: DINOVariant, input_size: int, feature_layers: tuple[int, ...], position_radius: int, spatial_weight: float, density_neighbors: int) -> AnomalyDINO` ¶

Instantiate and configure the AnomalyDINO model with frozen feature extractor.

Parameters:

Name	Type	Description	Default
`num_neighbors`	`int`	Nearest neighbors count for memory bank lookup.	required
`encoder_name`	`str`	Identifier for the backbone encoder.	required
`use_masking`	`bool`	Whether foreground PCA patch masking is active.	required
`model_generation`	`Literal['dinov2', 'dinov3']`	Architectural generation ('dinov2' or 'dinov3').	required
`variant`	`DINOVariant`	Model variant ('baseline' or 'enhanced').	required
`input_size`	`int`	Canonical square input dimension.	required
`feature_layers`	`tuple[int, ...]`	Block indices for enhanced multi-layer feature extraction.	required
`position_radius`	`int`	Position search radius for enhanced spatial matching.	required
`spatial_weight`	`float`	Weight penalty for off-center matches.	required
`density_neighbors`	`int`	Neighbor count for density estimation.	required

Returns:

Type	Description
`AnomalyDINO`	Configured AnomalyDINO instance with gradient updates disabled on encoder.

Source code in app/pipelines/modelling/dino/engine.py

def instantiate_dino_model(
    num_neighbors: int,
    encoder_name: str,
    use_masking: bool,
    model_generation: Literal["dinov2", "dinov3"],
    variant: DINOVariant,
    input_size: int,
    feature_layers: tuple[int, ...],
    position_radius: int,
    spatial_weight: float,
    density_neighbors: int,
) -> AnomalyDINO:
    """Instantiate and configure the AnomalyDINO model with frozen feature extractor.

    Args:
        num_neighbors: Nearest neighbors count for memory bank lookup.
        encoder_name: Identifier for the backbone encoder.
        use_masking: Whether foreground PCA patch masking is active.
        model_generation: Architectural generation ('dinov2' or 'dinov3').
        variant: Model variant ('baseline' or 'enhanced').
        input_size: Canonical square input dimension.
        feature_layers: Block indices for enhanced multi-layer feature extraction.
        position_radius: Position search radius for enhanced spatial matching.
        spatial_weight: Weight penalty for off-center matches.
        density_neighbors: Neighbor count for density estimation.

    Returns:
        Configured AnomalyDINO instance with gradient updates disabled on encoder.
    """
    model_kwargs: dict[str, Any] = {
        "num_neighbours": num_neighbors,
        "encoder_name": encoder_name,
        "masking": use_masking,
        "coreset_subsampling": False,
        "post_processor": False,
        "evaluator": False,
        "visualizer": RawScoreImageVisualizer(),
    }
    if model_generation == "dinov3":
        model_kwargs["pre_processor"] = AnomalyDINO.configure_pre_processor((input_size, input_size))

    model = AnomalyDINO(**model_kwargs)
    if variant == "enhanced":
        model.model = EnhancedAnomalyDINOModel(
            num_neighbours=num_neighbors,
            encoder_name=encoder_name,
            masking=use_masking,
            feature_layers=feature_layers,
            position_radius=position_radius,
            spatial_weight=spatial_weight,
            density_neighbours=density_neighbors,
        )
    model.model.feature_encoder.requires_grad_(False)
    return model

`load_completed_category_result(base_dir: Path, category: str, model_hash: str, run_heatmap: bool, load_heatmaps: bool = True) -> BaselineResult | None` ¶

Load a complete category result without retaining saved heatmap pixels.

Parameters:

Name	Type	Description	Default
`base_dir`	`Path`	Directory containing candidate evaluation artifacts.	required
`category`	`str`	MVTec category expected in the metadata.	required
`model_hash`	`str`	Unique SHA-256 fingerprint expected in the metadata.	required
`run_heatmap`	`bool`	Whether heatmap overlays are required to declare completion.	required
`load_heatmaps`	`bool`	Whether to deserialize heatmap pixels into the returned result.	`True`

Returns:

Type	Description
`BaselineResult \| None`	Reconstituted BaselineResult if complete artifacts exist, or None.

Source code in app/pipelines/modelling/dino/artifacts.py

def load_completed_category_result(
    base_dir: Path,
    category: str,
    model_hash: str,
    run_heatmap: bool,
    load_heatmaps: bool = True,
) -> BaselineResult | None:
    """Load a complete category result without retaining saved heatmap pixels.

    Args:
        base_dir: Directory containing candidate evaluation artifacts.
        category: MVTec category expected in the metadata.
        model_hash: Unique SHA-256 fingerprint expected in the metadata.
        run_heatmap: Whether heatmap overlays are required to declare completion.
        load_heatmaps: Whether to deserialize heatmap pixels into the returned result.

    Returns:
        Reconstituted BaselineResult if complete artifacts exist, or None.
    """
    metadata_path = base_dir / "metadata.json"
    required_artifacts = (base_dir / "image_metrics.npz", base_dir / "pixel_metrics.npz")
    if not metadata_path.is_file() or not all(path.is_file() for path in required_artifacts):
        return None

    try:
        metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
        if metadata.get("category") != category or metadata.get("hash") != model_hash:
            return None
        heatmap_path = metadata.get("heatmap_overlays_path")
        if run_heatmap and (not heatmap_path or not (base_dir / str(heatmap_path)).is_file()):
            return None

        heatmap_overlays = load_heatmap_overlays(base_dir / str(heatmap_path)) if heatmap_path and load_heatmaps else {}
        raw_results = metadata["raw_results"]
        result = format_results(
            test_results=[raw_results],
            category=category,
            base_dir=base_dir,
            manual_image_f1=float(metadata["image_f1"]),
            manual_pixel_f1=float(metadata["pixel_f1"]),
            manual_image_prec=float(metadata["image_precision"]),
            manual_image_rec=float(metadata["image_recall"]),
            img_threshold=float(metadata["image_threshold"]),
            pixel_threshold=float(metadata["pixel_threshold"]),
            pixel_auroc=float(metadata["pixel_auroc"]),
            pixel_aupimo=float(metadata["pixel_aupimo"]),
            anomaly_map_min=float(metadata["anomaly_map_min"]),
            anomaly_map_max=float(metadata["anomaly_map_max"]),
            anomaly_map_range=float(metadata["anomaly_map_range"]),
            heatmap_overlays=heatmap_overlays,
            anomalous_indices=list(metadata.get("anomalous_indices", [])),
            preprocessing_steps=list(metadata.get("preprocessing_steps", [])),
            hyperparameters=dict(metadata.get("hyperparameters", {})),
            dataset_split=dict(metadata.get("dataset_split", {})),
            model_hash=model_hash,
            metadata=metadata,
            true_positives=int(metadata["true_positives"]),
            false_positives=int(metadata["false_positives"]),
            false_negatives=int(metadata["false_negatives"]),
            true_negatives=int(metadata["true_negatives"]),
        )
        result["image_level"]["average_precision"] = float(metadata["image_average_precision"])
        return result
    except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError) as exc:
        logger.warning("Ignoring incomplete DINO artifacts in %s: %s", base_dir, exc)
        return None

`persist_dino_artifacts_and_format(category: str, base_dir: Path, model_hash: str, configuration_hash: str, model_generation: Literal['dinov2', 'dinov3'], variant: DINOVariant, encoder_name: str, num_neighbors: int, masking: MaskingMode, use_masking: bool, raw_prep_list: list[dict[str, Any]], hyperparameters: dict[str, Any], split_info: dict[str, Any], artifacts: EvaluationArtifacts, image_average_precision: float, fpr_limit: float, model_seed: int) -> BaselineResult` ¶

Serialize DINO run metadata and format standardized baseline result dictionary.

Parameters:

Name	Type	Description	Default
`category`	`str`	MVTec category evaluated.	required
`base_dir`	`Path`	Base directory where artifacts are saved.	required
`model_hash`	`str`	Unique run hash string.	required
`configuration_hash`	`str`	Stable hash for equivalent model configurations.	required
`model_generation`	`Literal['dinov2', 'dinov3']`	Encoder architecture family.	required
`variant`	`DINOVariant`	Scorer variant ('baseline' or 'enhanced').	required
`encoder_name`	`str`	Model backbone identifier.	required
`num_neighbors`	`int`	Nearest neighbor count.	required
`masking`	`MaskingMode`	Foreground masking policy mode.	required
`use_masking`	`bool`	Evaluated boolean decision for masking.	required
`raw_prep_list`	`list[dict[str, Any]]`	Normalized preprocessing configuration list.	required
`hyperparameters`	`dict[str, Any]`	Model hyperparameter dictionary.	required
`split_info`	`dict[str, Any]`	Dataset partition sample counts.	required
`artifacts`	`EvaluationArtifacts`	Structured EvaluationArtifacts container.	required
`image_average_precision`	`float`	Computed area under Precision-Recall curve.	required
`fpr_limit`	`float`	Maximum false-positive rate for AUPIMO integration.	required
`model_seed`	`int`	Deterministic random seed used for the run.	required

Returns:

Type	Description
`BaselineResult`	Structured BaselineResult dictionary complying with fair-eval-v1.

Source code in app/pipelines/modelling/dino/artifacts.py

def persist_dino_artifacts_and_format(
    category: str,
    base_dir: Path,
    model_hash: str,
    configuration_hash: str,
    model_generation: Literal["dinov2", "dinov3"],
    variant: DINOVariant,
    encoder_name: str,
    num_neighbors: int,
    masking: MaskingMode,
    use_masking: bool,
    raw_prep_list: list[dict[str, Any]],
    hyperparameters: dict[str, Any],
    split_info: dict[str, Any],
    artifacts: EvaluationArtifacts,
    image_average_precision: float,
    fpr_limit: float,
    model_seed: int,
) -> BaselineResult:
    """Serialize DINO run metadata and format standardized baseline result dictionary.

    Args:
        category: MVTec category evaluated.
        base_dir: Base directory where artifacts are saved.
        model_hash: Unique run hash string.
        configuration_hash: Stable hash for equivalent model configurations.
        model_generation: Encoder architecture family.
        variant: Scorer variant ('baseline' or 'enhanced').
        encoder_name: Model backbone identifier.
        num_neighbors: Nearest neighbor count.
        masking: Foreground masking policy mode.
        use_masking: Evaluated boolean decision for masking.
        raw_prep_list: Normalized preprocessing configuration list.
        hyperparameters: Model hyperparameter dictionary.
        split_info: Dataset partition sample counts.
        artifacts: Structured EvaluationArtifacts container.
        image_average_precision: Computed area under Precision-Recall curve.
        fpr_limit: Maximum false-positive rate for AUPIMO integration.
        model_seed: Deterministic random seed used for the run.

    Returns:
        Structured BaselineResult dictionary complying with fair-eval-v1.
    """
    raw_results: dict[str, float] = {
        "image_F1Score": artifacts.image_metrics.f1_score,
        "image_Precision": artifacts.image_metrics.precision,
        "image_Recall": artifacts.image_metrics.recall,
        "image_AUROC": artifacts.image_metrics.auroc,
        "image_AP": image_average_precision,
        "pixel_AUROC": artifacts.pixel_metrics.auroc,
        "pixel_F1Score": artifacts.pixel_metrics.f1_score,
        "pixel_AUPIMO": artifacts.pixel_metrics.aupimo,
    }
    print_anomalib_results_table(raw_results)

    heatmap_archive = save_heatmap_overlays(artifacts.heatmap_overlays, base_dir / "heatmap_overlays.npz")
    four_panel_dir = base_dir / "four_panel_images"
    metadata = {
        "hash": model_hash,
        "configuration_hash": configuration_hash,
        "model_type": f"{model_generation}_enhanced_knn" if variant == "enhanced" else f"{model_generation}_knn",
        "model_generation": model_generation,
        "category": category,
        "encoder_name": encoder_name,
        "num_neighbors": num_neighbors,
        "masking_mode": masking,
        "masking": use_masking,
        "coreset_subsampling": False,
        "variant": variant,
        "preprocessing_steps": raw_prep_list,
        "hyperparameters": hyperparameters,
        "dataset_split": split_info,
        "protocol": FAIR_EVALUATION_PROTOCOL,
        "threshold_source": "normal_validation",
        "image_threshold_quantile": PATCHCORE_IMAGE_THRESHOLD_QUANTILE,
        "pixel_threshold_quantile": PATCHCORE_PIXEL_THRESHOLD_QUANTILE,
        "model_seed": model_seed,
        "score_space": PATCHCORE_SCORE_SPACE,
        "pixel_metrics_version": PIXEL_METRICS_VERSION,
        "image_f1": artifacts.image_metrics.f1_score,
        "image_precision": artifacts.image_metrics.precision,
        "image_recall": artifacts.image_metrics.recall,
        "image_auroc": artifacts.image_metrics.auroc,
        "image_average_precision": image_average_precision,
        "pixel_f1": artifacts.pixel_metrics.f1_score,
        "pixel_auroc": artifacts.pixel_metrics.aupimo,
        "pixel_aupimo": artifacts.pixel_metrics.aupimo,
        "true_positives": artifacts.image_metrics.confusion.true_positives,
        "false_positives": artifacts.image_metrics.confusion.false_positives,
        "false_negatives": artifacts.image_metrics.confusion.false_negatives,
        "true_negatives": artifacts.image_metrics.confusion.true_negatives,
        "image_threshold": artifacts.thresholds.image,
        "pixel_threshold": artifacts.thresholds.pixel,
        "aupimo_fpr_bounds": list(AUPIMO_FPR_BOUNDS),
        "aupimo_num_thresholds": AUPIMO_NUM_THRESHOLDS,
        "canonical_height": CANONICAL_MAP_SIZE[0],
        "canonical_width": CANONICAL_MAP_SIZE[1],
        "anomaly_map_min": artifacts.pixel_metrics.anomaly_map_min,
        "anomaly_map_max": artifacts.pixel_metrics.anomaly_map_max,
        "anomaly_map_range": artifacts.pixel_metrics.anomaly_map_range,
        "heatmap_overlays_path": heatmap_archive.name if heatmap_archive is not None else None,
        "four_panel_images_path": four_panel_dir.name if four_panel_dir.is_dir() else None,
        "anomalous_indices": artifacts.anomalous_indices,
        "raw_results": raw_results,
        "timestamp": datetime.now(UTC).isoformat(),
    }
    (base_dir / "metadata.json").write_text(json.dumps(metadata, indent=4), encoding="utf-8")

    results = format_results(
        test_results=[raw_results],
        category=category,
        base_dir=base_dir,
        artifacts=artifacts,
        fpr_limit=fpr_limit,
        preprocessing_steps=raw_prep_list,
        hyperparameters=hyperparameters,
        dataset_split=split_info,
        model_hash=model_hash,
        metadata=metadata,
    )
    results["image_level"]["average_precision"] = image_average_precision
    return results

`release_accelerator_memory() -> None` ¶

Collect cyclic trainer state and release unused CUDA allocations.

Source code in app/pipelines/modelling/dino/engine.py

def release_accelerator_memory() -> None:
    """Collect cyclic trainer state and release unused CUDA allocations."""
    gc.collect()
    try:
        import torch

        if torch.cuda.is_available():
            torch.cuda.empty_cache()
    except ImportError:
        pass

`resolve_masking(masking: MaskingMode, category: str) -> bool` ¶

Resolve an explicit or published full-shot AnomalyDINO masking policy.

Parameters:

Name	Type	Description	Default
`masking`	`MaskingMode`	Active masking policy mode ('off', 'on', or 'published').	required
`category`	`str`	MVTec object or texture category name.	required

Returns:

Type	Description
`bool`	True if foreground PCA masking should be applied, False otherwise.

Raises:

Type	Description
`ValueError`	If masking mode is invalid.

Source code in app/pipelines/modelling/dino/engine.py

def resolve_masking(masking: MaskingMode, category: str) -> bool:
    """Resolve an explicit or published full-shot AnomalyDINO masking policy.

    Args:
        masking: Active masking policy mode ('off', 'on', or 'published').
        category: MVTec object or texture category name.

    Returns:
        True if foreground PCA masking should be applied, False otherwise.

    Raises:
        ValueError: If masking mode is invalid.
    """
    if masking == "off":
        return False
    if masking == "on":
        return True
    if masking == "published":
        return category in ANOMALY_DINO_MASKED_CATEGORIES
    raise ValueError("masking must be one of: off, on, published")

run_dino_all_categories(data_root: Path | str, pipeline: list[dict[str, Any]] | PreprocessingPipeline | None, fpr_limit: float, encoder_name: str, num_neighbors: int, masking: MaskingMode, run_heatmap: bool, registry_base: Path | str, model_seed: int, variant: DINOVariant, feature_layers: tuple[int, ...], position_radius: int, spatial_weight: float, density_neighbors: int, model_generation: Literal['dinov2', 'dinov3'], model_name: str, input_size: int, patch_size: int, batch_size: int, reuse_complete: bool = False, save_summary_files: bool = False) -> AllCategoriesResult ¶

Orchestrate sequential evaluation across all canonical MVTec categories.

Parameters:

Name	Type	Description	Default
`data_root`	`Path \| str`	Root directory of MVTec AD.	required
`pipeline`	`list[dict[str, Any]] \| PreprocessingPipeline \| None`	Preprocessing pipeline or configuration list.	required
`fpr_limit`	`float`	Fixed fair-eval AUPIMO FPR limit.	required
`encoder_name`	`str`	Backbone encoder identifier.	required
`num_neighbors`	`int`	Nearest neighbor count.	required
`masking`	`MaskingMode`	Foreground masking policy mode.	required
`run_heatmap`	`bool`	Whether to render test-image overlays.	required
`registry_base`	`Path \| str`	Target artifact directory.	required
`model_seed`	`int`	Deterministic random seed.	required
`variant`	`DINOVariant`	Scorer variant ('baseline' or 'enhanced').	required
`feature_layers`	`tuple[int, ...]`	Multi-layer indices for enhanced scoring.	required
`position_radius`	`int`	Spatial search radius.	required
`spatial_weight`	`float`	Spatial distance penalty.	required
`density_neighbors`	`int`	Neighbor count for density estimation.	required
`model_generation`	`Literal['dinov2', 'dinov3']`	Architectural family ('dinov2' or 'dinov3').	required
`model_name`	`str`	Label for logs and summaries.	required
`input_size`	`int`	Square model input dimension.	required
`patch_size`	`int`	Encoder patch dimension.	required
`batch_size`	`int`	DataLoader batch size.	required
`reuse_complete`	`bool`	Reuse complete artifacts for each category.	`False`
`save_summary_files`	`bool`	If True, writes summary.json and category_metrics.csv.	`False`

Returns:

Type	Description
`AllCategoriesResult`	AllCategoriesResult containing per-category results and macro averages.

Source code in app/pipelines/modelling/dino/engine.py

def run_dino_all_categories(
    data_root: Path | str,
    pipeline: list[dict[str, Any]] | PreprocessingPipeline | None,
    fpr_limit: float,
    encoder_name: str,
    num_neighbors: int,
    masking: MaskingMode,
    run_heatmap: bool,
    registry_base: Path | str,
    model_seed: int,
    variant: DINOVariant,
    feature_layers: tuple[int, ...],
    position_radius: int,
    spatial_weight: float,
    density_neighbors: int,
    model_generation: Literal["dinov2", "dinov3"],
    model_name: str,
    input_size: int,
    patch_size: int,
    batch_size: int,
    reuse_complete: bool = False,
    save_summary_files: bool = False,
) -> AllCategoriesResult:
    """Orchestrate sequential evaluation across all canonical MVTec categories.

    Args:
        data_root: Root directory of MVTec AD.
        pipeline: Preprocessing pipeline or configuration list.
        fpr_limit: Fixed fair-eval AUPIMO FPR limit.
        encoder_name: Backbone encoder identifier.
        num_neighbors: Nearest neighbor count.
        masking: Foreground masking policy mode.
        run_heatmap: Whether to render test-image overlays.
        registry_base: Target artifact directory.
        model_seed: Deterministic random seed.
        variant: Scorer variant ('baseline' or 'enhanced').
        feature_layers: Multi-layer indices for enhanced scoring.
        position_radius: Spatial search radius.
        spatial_weight: Spatial distance penalty.
        density_neighbors: Neighbor count for density estimation.
        model_generation: Architectural family ('dinov2' or 'dinov3').
        model_name: Label for logs and summaries.
        input_size: Square model input dimension.
        patch_size: Encoder patch dimension.
        batch_size: DataLoader batch size.
        reuse_complete: Reuse complete artifacts for each category.
        save_summary_files: If True, writes summary.json and category_metrics.csv.

    Returns:
        AllCategoriesResult containing per-category results and macro averages.
    """
    manifest = build_mvtec_manifest(data_root)
    category_results: dict[str, BaselineResult] = {}
    for name in MVTEC_CATEGORIES:
        try:
            result = run_dino_category(
                data_root=data_root,
                category=name,
                pipeline=pipeline,
                fpr_limit=fpr_limit,
                encoder_name=encoder_name,
                num_neighbors=num_neighbors,
                masking=masking,
                run_heatmap=run_heatmap,
                registry_base=registry_base,
                model_seed=model_seed,
                reuse_complete=reuse_complete,
                variant=variant,
                feature_layers=feature_layers,
                position_radius=position_radius,
                spatial_weight=spatial_weight,
                density_neighbors=density_neighbors,
                model_generation=model_generation,
                model_name=model_name,
                input_size=input_size,
                patch_size=patch_size,
                batch_size=batch_size,
                manifest=manifest,
            )
            result["heatmap_overlays"] = {}
            category_results[name] = result
            del result
        finally:
            release_accelerator_memory()

    macro_average = compute_macro_average(category_results)
    if save_summary_files:
        save_all_category_summary(
            registry_base=registry_base,
            encoder_name=encoder_name,
            masking=masking,
            num_neighbors=num_neighbors,
            category_results=category_results,
            macro_average=macro_average,
            model_name=model_name,
        )
    logger.info("%s all-category macro averages: %s", model_name, macro_average)
    return {
        "category": "all",
        "masking_mode": masking,
        "categories": category_results,
        "macro_average": macro_average,
    }

run_dino_category(data_root: Path | str = 'data/raw/mvtec_ad', category: str = 'bottle', pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None, fpr_limit: float = 0.0001, encoder_name: str = 'vit_small_patch14_dinov2', num_neighbors: int = 1, masking: MaskingMode = 'published', run_heatmap: bool = False, preprocessing_steps: list[dict[str, Any]] | None = None, registry_base: Path | str = 'data/models/dinov2', model_seed: int = PATCHCORE_MODEL_SEED, reuse_complete: bool = False, variant: DINOVariant = 'baseline', feature_layers: tuple[int, ...] = (8, 10, 11), position_radius: int = 1, spatial_weight: float = 0.05, density_neighbors: int = 5, model_generation: Literal['dinov2', 'dinov3'] = 'dinov2', model_name: str = 'DINOv2', input_size: int = 252, patch_size: int = 14, batch_size: int = 4, manifest: pd.DataFrame | None = None) -> BaselineResult ¶

Evaluate frozen DINO patch tokens with a normal-only nearest-neighbour bank.

Parameters:

Name	Type	Description	Default
`data_root`	`Path \| str`	Root directory of MVTec AD.	`'data/raw/mvtec_ad'`
`category`	`str`	MVTec category to evaluate.	`'bottle'`
`pipeline`	`list[dict[str, Any]] \| PreprocessingPipeline \| None`	Optional preprocessing pipeline or configuration list.	`None`
`fpr_limit`	`float`	Upper AUPIMO false-positive-rate bound fixed by the fair protocol.	`0.0001`
`encoder_name`	`str`	Pretrained DINO encoder exposed by Anomalib/timm.	`'vit_small_patch14_dinov2'`
`num_neighbors`	`int`	Number of normal patch neighbours averaged per patch.	`1`
`masking`	`MaskingMode`	PCA foreground-mask policy: disabled, enabled, or published.	`'published'`
`run_heatmap`	`bool`	Whether to render overlays for anomalous test images.	`False`
`preprocessing_steps`	`list[dict[str, Any]] \| None`	Legacy parameter preserved for backward compatibility.	`None`
`registry_base`	`Path \| str`	Directory in which evaluation artifacts are written.	`'data/models/dinov2'`
`model_seed`	`int`	Shared deterministic model and data-loader seed.	`PATCHCORE_MODEL_SEED`
`reuse_complete`	`bool`	Reuse complete artifacts for this exact configuration.	`False`
`variant`	`DINOVariant`	Stock final-block scorer or enhanced multi-layer scorer.	`'baseline'`
`feature_layers`	`tuple[int, ...]`	Transformer block indices used by the enhanced scorer.	`(8, 10, 11)`
`position_radius`	`int`	Patch-grid search radius used by the enhanced scorer.	`1`
`spatial_weight`	`float`	Spatial-distance penalty used by the enhanced scorer.	`0.05`
`density_neighbors`	`int`	Normal neighbours used to estimate local density.	`5`
`model_generation`	`Literal['dinov2', 'dinov3']`	Encoder family ('dinov2' or 'dinov3').	`'dinov2'`
`model_name`	`str`	Human-readable model label used in logs.	`'DINOv2'`
`input_size`	`int`	Square model input size recorded in the run metadata.	`252`
`patch_size`	`int`	Encoder patch size recorded in the run metadata.	`14`
`batch_size`	`int`	Batch size for training datamodule.	`4`
`manifest`	`DataFrame \| None`	Optional prebuilt dataset manifest for all-category orchestration.	`None`

Returns:

Type	Description
`BaselineResult`	Results using the standardized BaselineResult schema.

Raises:

Type	Description
`ValueError`	If validation bounds or encoder specifications are invalid.
`RuntimeError`	If memory bank fitting produces zero normal patches.

Source code in app/pipelines/modelling/dino/engine.py

def run_dino_category(
    data_root: Path | str = "data/raw/mvtec_ad",
    category: str = "bottle",
    pipeline: list[dict[str, Any]] | PreprocessingPipeline | None = None,
    fpr_limit: float = 1e-4,
    encoder_name: str = "vit_small_patch14_dinov2",
    num_neighbors: int = 1,
    masking: MaskingMode = "published",
    run_heatmap: bool = False,
    preprocessing_steps: list[dict[str, Any]] | None = None,
    registry_base: Path | str = "data/models/dinov2",
    model_seed: int = PATCHCORE_MODEL_SEED,
    reuse_complete: bool = False,
    variant: DINOVariant = "baseline",
    feature_layers: tuple[int, ...] = (8, 10, 11),
    position_radius: int = 1,
    spatial_weight: float = 0.05,
    density_neighbors: int = 5,
    model_generation: Literal["dinov2", "dinov3"] = "dinov2",
    model_name: str = "DINOv2",
    input_size: int = 252,
    patch_size: int = 14,
    batch_size: int = 4,
    manifest: pd.DataFrame | None = None,
) -> BaselineResult:
    """Evaluate frozen DINO patch tokens with a normal-only nearest-neighbour bank.

    Args:
        data_root: Root directory of MVTec AD.
        category: MVTec category to evaluate.
        pipeline: Optional preprocessing pipeline or configuration list.
        fpr_limit: Upper AUPIMO false-positive-rate bound fixed by the fair protocol.
        encoder_name: Pretrained DINO encoder exposed by Anomalib/timm.
        num_neighbors: Number of normal patch neighbours averaged per patch.
        masking: PCA foreground-mask policy: disabled, enabled, or published.
        run_heatmap: Whether to render overlays for anomalous test images.
        preprocessing_steps: Legacy parameter preserved for backward compatibility.
        registry_base: Directory in which evaluation artifacts are written.
        model_seed: Shared deterministic model and data-loader seed.
        reuse_complete: Reuse complete artifacts for this exact configuration.
        variant: Stock final-block scorer or enhanced multi-layer scorer.
        feature_layers: Transformer block indices used by the enhanced scorer.
        position_radius: Patch-grid search radius used by the enhanced scorer.
        spatial_weight: Spatial-distance penalty used by the enhanced scorer.
        density_neighbors: Normal neighbours used to estimate local density.
        model_generation: Encoder family ('dinov2' or 'dinov3').
        model_name: Human-readable model label used in logs.
        input_size: Square model input size recorded in the run metadata.
        patch_size: Encoder patch size recorded in the run metadata.
        batch_size: Batch size for training datamodule.
        manifest: Optional prebuilt dataset manifest for all-category orchestration.

    Returns:
        Results using the standardized BaselineResult schema.

    Raises:
        ValueError: If validation bounds or encoder specifications are invalid.
        RuntimeError: If memory bank fitting produces zero normal patches.
    """
    _validate_dino_parameters(fpr_limit, num_neighbors, model_generation, encoder_name, model_name)
    use_masking = resolve_masking(masking, category)

    steps_config = pipeline if pipeline is not None else preprocessing_steps
    if isinstance(steps_config, PreprocessingPipeline):
        proc_pipeline = steps_config
        raw_prep_list: list[dict[str, Any]] = []
    else:
        proc_pipeline = build_pipeline_from_configs(steps_config)
        raw_prep_list = normalize_preprocessing_steps(steps_config)

    active_manifest = manifest if manifest is not None else build_mvtec_manifest(data_root)
    fair_split = build_fair_evaluation_split(active_manifest, category)
    cache_evidence = {
        **fair_split.evidence(),
        **fair_metric_evidence(),
        "model_seed": model_seed,
        "score_space": PATCHCORE_SCORE_SPACE,
        "image_threshold_quantile": PATCHCORE_IMAGE_THRESHOLD_QUANTILE,
        "pixel_threshold_quantile": PATCHCORE_PIXEL_THRESHOLD_QUANTILE,
    }

    _identity, configuration_hash, enhanced_scorer = build_dino_identity_and_hash(
        category=category,
        encoder_name=encoder_name,
        num_neighbors=num_neighbors,
        masking=masking,
        use_masking=use_masking,
        raw_prep_list=raw_prep_list,
        cache_evidence=cache_evidence,
        model_generation=model_generation,
        variant=variant,
        feature_layers=feature_layers,
        position_radius=position_radius,
        spatial_weight=spatial_weight,
        density_neighbors=density_neighbors,
    )

    active_hash = configuration_hash
    base_dir = Path(registry_base) / active_hash
    if reuse_complete:
        completed_result = load_completed_category_result(
            base_dir, category, active_hash, run_heatmap, load_heatmaps=False
        )
        if completed_result is not None:
            logger.info("Reusing complete %s result for %s (Hash: %s)", model_name, category, active_hash)
            return completed_result

        _rotate_dino_cache(base_dir, Path(registry_base), configuration_hash)
    base_dir.mkdir(parents=True)

    datamodule = MVTecAD(
        root=data_root,
        category=category,
        train_batch_size=batch_size,
        eval_batch_size=batch_size,
        val_split_mode="none",
    )
    transform_adapter = PreprocessingTransformAdapter(proc_pipeline)
    configure_anomalib_partitions(
        datamodule,
        fair_split,
        transform_adapter if len(proc_pipeline) > 0 else None,
    )

    seed_anomalib_run(model_seed)
    model = instantiate_dino_model(
        num_neighbors=num_neighbors,
        encoder_name=encoder_name,
        use_masking=use_masking,
        model_generation=model_generation,
        variant=variant,
        input_size=input_size,
        feature_layers=feature_layers,
        position_radius=position_radius,
        spatial_weight=spatial_weight,
        density_neighbors=density_neighbors,
    )
    if isinstance(model.visualizer, RawScoreImageVisualizer):
        model.visualizer.output_dir = base_dir / "four_panel_images"

    engine = Engine(accelerator="auto", devices=1, deterministic=True)
    train_dataloader = datamodule.train_dataloader()
    validation_dataloader = datamodule.val_dataloader()
    test_dataloader = datamodule.test_dataloader()

    logger.info("Building %s normal patch bank for %s (Hash: %s)...", model_name, category, active_hash)
    engine.fit(model, train_dataloaders=train_dataloader)
    memory_bank = getattr(model.model, "memory_bank", None)
    if memory_bank is not None and hasattr(memory_bank, "numel") and memory_bank.numel() == 0:
        raise RuntimeError(
            f"{model_name} fitting produced an empty memory bank for category '{category}'. "
            "Check that the fitting loader contains normal images and that foreground masking retains patches."
        )

    artifacts = EvaluationArtifacts.from_tuple(
        extract_and_save_pr_metrics(
            engine,
            model,
            validation_dataloader,
            test_dataloader,
            base_dir,
            run_heatmap,
            model_name=model_name,
        )
    )

    with np.load(base_dir / "image_metrics.npz", allow_pickle=False) as image_metrics:
        precision_curve = np.asarray(image_metrics["precision"], dtype=np.float64)
        recall_curve = np.asarray(image_metrics["recall"], dtype=np.float64)
    image_average_precision = float(-np.sum(np.diff(recall_curve) * precision_curve[:-1]))

    split_info = {
        **cache_evidence,
        "test_normal": int((fair_split.test["is_anomaly"] == 0).sum()),
        "test_anomalous": int(fair_split.test["is_anomaly"].astype(bool).sum()),
    }
    hyperparameters = {
        "encoder_name": encoder_name,
        "input_size": input_size,
        "patch_size": patch_size,
        "num_neighbors": num_neighbors,
        "masking_mode": masking,
        "masking": use_masking,
        "coreset_subsampling": False,
        "train_batch_size": batch_size,
        "eval_batch_size": batch_size,
        "model_seed": model_seed,
        "score_space": PATCHCORE_SCORE_SPACE,
        "variant": variant,
    }
    if variant == "enhanced":
        hyperparameters.update(enhanced_scorer)

    return persist_dino_artifacts_and_format(
        category=category,
        base_dir=base_dir,
        model_hash=active_hash,
        configuration_hash=configuration_hash,
        model_generation=model_generation,
        variant=variant,
        encoder_name=encoder_name,
        num_neighbors=num_neighbors,
        masking=masking,
        use_masking=use_masking,
        raw_prep_list=raw_prep_list,
        hyperparameters=hyperparameters,
        split_info=split_info,
        artifacts=artifacts,
        image_average_precision=image_average_precision,
        fpr_limit=fpr_limit,
        model_seed=model_seed,
    )

`save_all_category_summary(registry_base: Path | str, encoder_name: str, masking: MaskingMode, num_neighbors: int, category_results: dict[str, BaselineResult], macro_average: dict[str, float], model_name: str = 'DINO') -> None` ¶

Persist compact machine-readable summaries beside category artifacts.

Parameters:

Name	Type	Description	Default
`registry_base`	`Path \| str`	Target directory where summary files are placed.	required
`encoder_name`	`str`	Name of the encoder backbone used.	required
`masking`	`MaskingMode`	Foreground masking policy mode.	required
`num_neighbors`	`int`	Nearest neighbor count.	required
`category_results`	`dict[str, BaselineResult]`	Mapping of category names to their BaselineResults.	required
`macro_average`	`dict[str, float]`	Unweighted metric averages across categories.	required
`model_name`	`str`	Label used in the written JSON summary.	`'DINO'`

Source code in app/pipelines/modelling/dino/artifacts.py

def save_all_category_summary(
    registry_base: Path | str,
    encoder_name: str,
    masking: MaskingMode,
    num_neighbors: int,
    category_results: dict[str, BaselineResult],
    macro_average: dict[str, float],
    model_name: str = "DINO",
) -> None:
    """Persist compact machine-readable summaries beside category artifacts.

    Args:
        registry_base: Target directory where summary files are placed.
        encoder_name: Name of the encoder backbone used.
        masking: Foreground masking policy mode.
        num_neighbors: Nearest neighbor count.
        category_results: Mapping of category names to their BaselineResults.
        macro_average: Unweighted metric averages across categories.
        model_name: Label used in the written JSON summary.
    """
    output_dir = Path(registry_base)
    output_dir.mkdir(parents=True, exist_ok=True)
    metric_keys = tuple(macro_average)
    category_metrics: dict[str, dict[str, float]] = {}
    for name, result in category_results.items():
        category_metrics[name] = {
            "image_f1": float(result["image_level"]["f1_score"]),
            "image_recall": float(result["image_level"]["recall"]),
            "image_precision": float(result["image_level"]["precision"]),
            "image_auroc": float(result["image_level"]["auroc"]),
            "image_average_precision": float(result["image_level"]["average_precision"]),
            "pixel_f1": float(result["pixel_level"]["f1_score"]),
            "pixel_auroc": float(result["pixel_level"]["auroc"]),
            "pixel_aupimo": float(result["pixel_level"]["aupimo"]),
        }

    summary = {
        "model": model_name,
        "variant": "baseline_single_layer_1nn" if num_neighbors == 1 else f"baseline_single_layer_{num_neighbors}nn",
        "encoder_name": encoder_name,
        "masking_mode": masking,
        "categories": len(category_results),
        "protocol": FAIR_EVALUATION_PROTOCOL,
        "macro_average": macro_average,
        "category_metrics": category_metrics,
    }
    (output_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")

    with (output_dir / "category_metrics.csv").open("w", newline="", encoding="utf-8") as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=("category", *metric_keys))
        writer.writeheader()
        for name, metrics in category_metrics.items():
            writer.writerow({"category": name, **metrics})

Preprocessing Subpackage¶

`app.pipelines.preprocessing.factory` ¶

Preprocessing factory module for building preprocessing pipelines.

`build_pipeline_from_configs(configs: list[dict[str, Any]] | None) -> PreprocessingPipeline` ¶

Build a PreprocessingPipeline from a list of dict configs.

Example input

[ {"name": "clahe", "params": {"clip_limit": 3.0}}, {"name": "gaussian_blur", "params": {"kernel_size": 3}} ]

Parameters:

Name	Type	Description	Default
`configs`	`list[dict[str, Any]] \| None`	List of preprocessing step configurations.	required

Returns:

Type	Description
`PreprocessingPipeline`	PreprocessingPipeline with steps added from configs.

Source code in app/pipelines/preprocessing/factory.py

def build_pipeline_from_configs(
    configs: list[dict[str, Any]] | None,
) -> PreprocessingPipeline:
    """Build a PreprocessingPipeline from a list of dict configs.

    Example input:
        [
            {"name": "clahe", "params": {"clip_limit": 3.0}},
            {"name": "gaussian_blur", "params": {"kernel_size": 3}}
        ]

    Args:
        configs: List of preprocessing step configurations.

    Returns:
        PreprocessingPipeline with steps added from configs.
    """
    pipeline = PreprocessingPipeline()
    if not configs:
        return pipeline

    for config in configs:
        step_name = config.get("name")
        params = config.get("params", {})
        if step_name in STEP_REGISTRY:
            step_cls = STEP_REGISTRY[step_name]
            pipeline.add_step(step_cls(**params))

    return pipeline

`normalize_preprocessing_steps(steps: list[dict[str, Any]] | None) -> list[dict[str, Any]]` ¶

Normalize a list of preprocessing step dicts for deterministic hashing and registry lookups.

Sorts parameter keys alphabetically and strips extraneous or unhashable attributes.

Parameters:

Name	Type	Description	Default
`steps`	`list[dict[str, Any]] \| None`	List of raw preprocessing step dictionaries (e.g. `[{"name": "clahe", "params": {...}}]`).	required

Returns:

Type	Description
`list[dict[str, Any]]`	List of normalized step dictionaries with sorted param dictionaries.

Source code in app/pipelines/preprocessing/factory.py

def normalize_preprocessing_steps(
    steps: list[dict[str, Any]] | None,
) -> list[dict[str, Any]]:
    """Normalize a list of preprocessing step dicts for deterministic hashing and registry lookups.

    Sorts parameter keys alphabetically and strips extraneous or unhashable attributes.

    Args:
        steps: List of raw preprocessing step dictionaries (e.g. `[{"name": "clahe", "params": {...}}]`).

    Returns:
        List of normalized step dictionaries with sorted param dictionaries.
    """
    if not steps:
        return []

    normalized: list[dict[str, Any]] = []
    for step in steps:
        if isinstance(step, dict):
            item: dict[str, Any] = {"name": step.get("name")}
            params = step.get("params")
            if isinstance(params, dict):
                item["params"] = dict(sorted(params.items()))
            normalized.append(item)
    return normalized

`app.pipelines.preprocessing.augmentation` ¶

Category-aware data augmentation for the Keras CAE anomaly detection pipeline.

Why Does Augmentation Strategy Matter for Anomaly Detection?¶

In industrial anomaly detection, the autoencoder is trained exclusively on defect-free normal images. The goal of augmentation is NOT to help the model generalise to new defect classes (that would be wrong), but to:

Prevent overfitting to the exact photographic conditions of the training set (lighting angles, minor camera vibration, batch-to-batch variation).
Make the model robust to permissible natural variance (e.g., slightly different grain orientation in wood), while still flagging genuine defects as anomalous.

The Critical Distinction: Textures vs. Rigid Objects¶

This is the most important design decision in augmentation:

Texture categories (wood, carpet, leather, tile, grid): These materials have spatial invariance - the statistical pattern of wood grain looks essentially the same whether you rotate it 90 degrees or not. Therefore, heavy geometric augmentations (random rotations, flips, scale jitter) are very effective. They teach the model "what does normal wood look like from any direction?".

Rigid object categories (transistor, pill, screw, capsule, metal_nut, bolt): These objects are directionally aligned on the conveyor belt or inspection jig. A transistor always arrives with its legs pointing down. If you rotate it 90 degrees during training, the model learns that an upside-down transistor is normal - which destroys the anomaly detection capability for orientation defects entirely. For these, only light colour/intensity augmentations are safe.

Module Contents¶

TEXTURE_CATEGORIES: Set of MVTec categories that are textures.
OBJECT_CATEGORIES: Set of MVTec categories that are rigid objects.
TextureAugmenter: Heavy augmentation pipeline for textures.
ObjectAugmenter: Light augmentation pipeline for rigid objects.
get_augmenter: Factory that returns the correct augmenter by category name.

`OBJECT_CATEGORIES: frozenset[str] = frozenset(MVTEC_OBJECT_CATEGORIES)` `module-attribute` ¶

Set representation of rigid object categories for fast O(1) membership checks.

`TEXTURE_CATEGORIES: frozenset[str] = frozenset(MVTEC_TEXTURE_CATEGORIES)` `module-attribute` ¶

Set representation of surface texture categories for fast O(1) membership checks.

`ObjectAugmenter` ¶

Light augmentation pipeline for directionally aligned rigid object categories.

These objects (e.g., transistors, pills, screws) are always positioned in a consistent orientation in the MVTec dataset. Applying rotations would teach the model that an upside-down transistor is "normal" - completely defeating anomaly detection for orientation-related defects.

Therefore, only photometric (colour/intensity) augmentations are applied: 1. Slight brightness jitter (±10%). 2. Slight contrast jitter (±10%). 3. Slight saturation jitter (±10%). 4. Light additive Gaussian noise (very small standard deviation, ±2% intensity).

Attributes:

Name	Type	Description
`brightness_range`		Tuple (min, max) brightness multiplier.
`contrast_range`		Tuple (min, max) contrast multiplier.
`saturation_range`		Tuple (min, max) colour saturation multiplier.
`noise_std`		Standard deviation of additive Gaussian noise (0.0 to 1.0 scale).

Source code in app/pipelines/preprocessing/augmentation.py

class ObjectAugmenter:
    """Light augmentation pipeline for directionally aligned rigid object categories.

    These objects (e.g., transistors, pills, screws) are always positioned in a
    consistent orientation in the MVTec dataset. Applying rotations would teach the
    model that an upside-down transistor is "normal" - completely defeating anomaly
    detection for orientation-related defects.

    Therefore, only **photometric** (colour/intensity) augmentations are applied:
    1. Slight brightness jitter (±10%).
    2. Slight contrast jitter (±10%).
    3. Slight saturation jitter (±10%).
    4. Light additive Gaussian noise (very small standard deviation, ±2% intensity).

    Attributes:
        brightness_range: Tuple (min, max) brightness multiplier.
        contrast_range: Tuple (min, max) contrast multiplier.
        saturation_range: Tuple (min, max) colour saturation multiplier.
        noise_std: Standard deviation of additive Gaussian noise (0.0 to 1.0 scale).
    """

    def __init__(
        self,
        brightness_range: tuple[float, float] = (0.9, 1.1),
        contrast_range: tuple[float, float] = (0.9, 1.1),
        saturation_range: tuple[float, float] = (0.9, 1.1),
        noise_std: float = 0.02,
    ) -> None:
        """Initialize the object augmenter with configurable photometric jitter.

        Args:
            brightness_range: (min, max) brightness multiplier. 1.0 = no change.
            contrast_range: (min, max) contrast multiplier. 1.0 = no change.
            saturation_range: (min, max) saturation multiplier. 1.0 = no change.
            noise_std: Standard deviation of Gaussian noise added to normalised [0,1] pixels.
        """
        self.brightness_range = brightness_range
        self.contrast_range = contrast_range
        self.saturation_range = saturation_range
        self.noise_std = noise_std

    def __call__(self, image: Image.Image) -> Image.Image:
        """Apply the light object augmentation pipeline to a single PIL image.

        Args:
            image: Input PIL Image in RGB mode.

        Returns:
            Augmented PIL Image in RGB mode, same size as input.
        """
        # 1. Slight brightness jitter
        brightness_factor = random.uniform(*self.brightness_range)
        image = ImageEnhance.Brightness(image).enhance(brightness_factor)

        # 2. Slight contrast jitter
        contrast_factor = random.uniform(*self.contrast_range)
        image = ImageEnhance.Contrast(image).enhance(contrast_factor)

        # 3. Slight colour saturation jitter
        saturation_factor = random.uniform(*self.saturation_range)
        image = ImageEnhance.Color(image).enhance(saturation_factor)

        # 4. Additive Gaussian noise (very light synthetic sensor noise)
        if self.noise_std > 0.0:
            image_array = np.array(image, dtype=np.float32) / 255.0
            noise = np.random.normal(loc=0.0, scale=self.noise_std, size=image_array.shape)
            image_array = np.clip(image_array + noise, 0.0, 1.0)
            image = Image.fromarray((image_array * 255).astype(np.uint8))

        return image

`call(image: Image.Image) -> Image.Image` ¶

Apply the light object augmentation pipeline to a single PIL image.

Parameters:

Name	Type	Description	Default
`image`	`Image`	Input PIL Image in RGB mode.	required

Returns:

Type	Description
`Image`	Augmented PIL Image in RGB mode, same size as input.

Source code in app/pipelines/preprocessing/augmentation.py

def __call__(self, image: Image.Image) -> Image.Image:
    """Apply the light object augmentation pipeline to a single PIL image.

    Args:
        image: Input PIL Image in RGB mode.

    Returns:
        Augmented PIL Image in RGB mode, same size as input.
    """
    # 1. Slight brightness jitter
    brightness_factor = random.uniform(*self.brightness_range)
    image = ImageEnhance.Brightness(image).enhance(brightness_factor)

    # 2. Slight contrast jitter
    contrast_factor = random.uniform(*self.contrast_range)
    image = ImageEnhance.Contrast(image).enhance(contrast_factor)

    # 3. Slight colour saturation jitter
    saturation_factor = random.uniform(*self.saturation_range)
    image = ImageEnhance.Color(image).enhance(saturation_factor)

    # 4. Additive Gaussian noise (very light synthetic sensor noise)
    if self.noise_std > 0.0:
        image_array = np.array(image, dtype=np.float32) / 255.0
        noise = np.random.normal(loc=0.0, scale=self.noise_std, size=image_array.shape)
        image_array = np.clip(image_array + noise, 0.0, 1.0)
        image = Image.fromarray((image_array * 255).astype(np.uint8))

    return image

`init(brightness_range: tuple[float, float] = (0.9, 1.1), contrast_range: tuple[float, float] = (0.9, 1.1), saturation_range: tuple[float, float] = (0.9, 1.1), noise_std: float = 0.02) -> None` ¶

Initialize the object augmenter with configurable photometric jitter.

Parameters:

Name	Type	Description	Default
`brightness_range`	`tuple[float, float]`	(min, max) brightness multiplier. 1.0 = no change.	`(0.9, 1.1)`
`contrast_range`	`tuple[float, float]`	(min, max) contrast multiplier. 1.0 = no change.	`(0.9, 1.1)`
`saturation_range`	`tuple[float, float]`	(min, max) saturation multiplier. 1.0 = no change.	`(0.9, 1.1)`
`noise_std`	`float`	Standard deviation of Gaussian noise added to normalised [0,1] pixels.	`0.02`

Source code in app/pipelines/preprocessing/augmentation.py

def __init__(
    self,
    brightness_range: tuple[float, float] = (0.9, 1.1),
    contrast_range: tuple[float, float] = (0.9, 1.1),
    saturation_range: tuple[float, float] = (0.9, 1.1),
    noise_std: float = 0.02,
) -> None:
    """Initialize the object augmenter with configurable photometric jitter.

    Args:
        brightness_range: (min, max) brightness multiplier. 1.0 = no change.
        contrast_range: (min, max) contrast multiplier. 1.0 = no change.
        saturation_range: (min, max) saturation multiplier. 1.0 = no change.
        noise_std: Standard deviation of Gaussian noise added to normalised [0,1] pixels.
    """
    self.brightness_range = brightness_range
    self.contrast_range = contrast_range
    self.saturation_range = saturation_range
    self.noise_std = noise_std

`TextureAugmenter` ¶

Heavy augmentation pipeline for spatially invariant texture categories.

Spatial invariance means the visual statistics of the material do not fundamentally change under rotation or reflection. Wood grain rotated 90 degrees still looks like normal wood - so we exploit this to generate more training variety.

Augmentations applied in random order: 1. Random 90°/180°/270° rotation (or no rotation). 2. Random horizontal flip. 3. Random vertical flip. 4. Random scale crop (zooms into 80-100% of the image, then resizes back). 5. Slight brightness jitter (±20% brightness variation). 6. Slight contrast jitter (±20% contrast variation).

Attributes:

Name	Type	Description
`brightness_range`		Tuple (min, max) multiplier for brightness jitter.
`contrast_range`		Tuple (min, max) multiplier for contrast jitter.
`scale_range`		Tuple (min_crop_fraction, max_crop_fraction) for scale jitter.

Source code in app/pipelines/preprocessing/augmentation.py

class TextureAugmenter:
    """Heavy augmentation pipeline for spatially invariant texture categories.

    Spatial invariance means the visual statistics of the material do not fundamentally
    change under rotation or reflection. Wood grain rotated 90 degrees still looks like
    normal wood - so we exploit this to generate more training variety.

    Augmentations applied in random order:
    1. Random 90°/180°/270° rotation (or no rotation).
    2. Random horizontal flip.
    3. Random vertical flip.
    4. Random scale crop (zooms into 80-100% of the image, then resizes back).
    5. Slight brightness jitter (±20% brightness variation).
    6. Slight contrast jitter (±20% contrast variation).

    Attributes:
        brightness_range: Tuple (min, max) multiplier for brightness jitter.
        contrast_range: Tuple (min, max) multiplier for contrast jitter.
        scale_range: Tuple (min_crop_fraction, max_crop_fraction) for scale jitter.
    """

    def __init__(
        self,
        brightness_range: tuple[float, float] = (0.8, 1.2),
        contrast_range: tuple[float, float] = (0.8, 1.2),
        scale_range: tuple[float, float] = (0.8, 1.0),
    ) -> None:
        """Initialize the texture augmenter with configurable jitter ranges.

        Args:
            brightness_range: (min, max) brightness multiplier. 1.0 = original.
            contrast_range: (min, max) contrast multiplier. 1.0 = original.
            scale_range: (min_fraction, max_fraction) of image area to crop before resize.
        """
        self.brightness_range = brightness_range
        self.contrast_range = contrast_range
        self.scale_range = scale_range

    def __call__(self, image: Image.Image) -> Image.Image:
        """Apply the full texture augmentation pipeline to a single PIL image.

        Args:
            image: Input PIL Image in RGB mode.

        Returns:
            Augmented PIL Image in RGB mode, same size as input.
        """
        original_size = image.size  # (width, height) in PIL convention

        # 1. Random 90-degree rotation (0, 90, 180, or 270 degrees)
        rotation_angle = random.choice([0, 90, 180, 270])
        if rotation_angle != 0:
            image = image.rotate(rotation_angle, expand=False)

        # 2. Random horizontal flip (50% probability)
        if random.random() < 0.5:
            image = image.transpose(Image.Transpose.FLIP_LEFT_RIGHT)

        # 3. Random vertical flip (50% probability)
        if random.random() < 0.5:
            image = image.transpose(Image.Transpose.FLIP_TOP_BOTTOM)

        # 4. Random scale jitter (crop a random sub-region and resize back)
        crop_fraction = random.uniform(*self.scale_range)
        if crop_fraction < 1.0:
            w, h = image.size
            crop_w = int(w * crop_fraction)
            crop_h = int(h * crop_fraction)
            left = random.randint(0, w - crop_w)
            top = random.randint(0, h - crop_h)
            image = image.crop((left, top, left + crop_w, top + crop_h))
            image = image.resize(original_size, Image.Resampling.BILINEAR)

        # 5. Random brightness jitter
        brightness_factor = random.uniform(*self.brightness_range)
        image = ImageEnhance.Brightness(image).enhance(brightness_factor)

        # 6. Random contrast jitter
        contrast_factor = random.uniform(*self.contrast_range)
        image = ImageEnhance.Contrast(image).enhance(contrast_factor)

        return image

`call(image: Image.Image) -> Image.Image` ¶

Apply the full texture augmentation pipeline to a single PIL image.

Parameters:

Name	Type	Description	Default
`image`	`Image`	Input PIL Image in RGB mode.	required

Returns:

Type	Description
`Image`	Augmented PIL Image in RGB mode, same size as input.

Source code in app/pipelines/preprocessing/augmentation.py

def __call__(self, image: Image.Image) -> Image.Image:
    """Apply the full texture augmentation pipeline to a single PIL image.

    Args:
        image: Input PIL Image in RGB mode.

    Returns:
        Augmented PIL Image in RGB mode, same size as input.
    """
    original_size = image.size  # (width, height) in PIL convention

    # 1. Random 90-degree rotation (0, 90, 180, or 270 degrees)
    rotation_angle = random.choice([0, 90, 180, 270])
    if rotation_angle != 0:
        image = image.rotate(rotation_angle, expand=False)

    # 2. Random horizontal flip (50% probability)
    if random.random() < 0.5:
        image = image.transpose(Image.Transpose.FLIP_LEFT_RIGHT)

    # 3. Random vertical flip (50% probability)
    if random.random() < 0.5:
        image = image.transpose(Image.Transpose.FLIP_TOP_BOTTOM)

    # 4. Random scale jitter (crop a random sub-region and resize back)
    crop_fraction = random.uniform(*self.scale_range)
    if crop_fraction < 1.0:
        w, h = image.size
        crop_w = int(w * crop_fraction)
        crop_h = int(h * crop_fraction)
        left = random.randint(0, w - crop_w)
        top = random.randint(0, h - crop_h)
        image = image.crop((left, top, left + crop_w, top + crop_h))
        image = image.resize(original_size, Image.Resampling.BILINEAR)

    # 5. Random brightness jitter
    brightness_factor = random.uniform(*self.brightness_range)
    image = ImageEnhance.Brightness(image).enhance(brightness_factor)

    # 6. Random contrast jitter
    contrast_factor = random.uniform(*self.contrast_range)
    image = ImageEnhance.Contrast(image).enhance(contrast_factor)

    return image

`init(brightness_range: tuple[float, float] = (0.8, 1.2), contrast_range: tuple[float, float] = (0.8, 1.2), scale_range: tuple[float, float] = (0.8, 1.0)) -> None` ¶

Initialize the texture augmenter with configurable jitter ranges.

Parameters:

Name	Type	Description	Default
`brightness_range`	`tuple[float, float]`	(min, max) brightness multiplier. 1.0 = original.	`(0.8, 1.2)`
`contrast_range`	`tuple[float, float]`	(min, max) contrast multiplier. 1.0 = original.	`(0.8, 1.2)`
`scale_range`	`tuple[float, float]`	(min_fraction, max_fraction) of image area to crop before resize.	`(0.8, 1.0)`

Source code in app/pipelines/preprocessing/augmentation.py

def __init__(
    self,
    brightness_range: tuple[float, float] = (0.8, 1.2),
    contrast_range: tuple[float, float] = (0.8, 1.2),
    scale_range: tuple[float, float] = (0.8, 1.0),
) -> None:
    """Initialize the texture augmenter with configurable jitter ranges.

    Args:
        brightness_range: (min, max) brightness multiplier. 1.0 = original.
        contrast_range: (min, max) contrast multiplier. 1.0 = original.
        scale_range: (min_fraction, max_fraction) of image area to crop before resize.
    """
    self.brightness_range = brightness_range
    self.contrast_range = contrast_range
    self.scale_range = scale_range

`augment_batch(images: np.ndarray, augmenter: TextureAugmenter | ObjectAugmenter) -> np.ndarray` ¶

Apply augmentation to a batch of numpy images.

Parameters:

Name	Type	Description	Default
`images`	`ndarray`	Batch of images as a numpy array of shape (N, H, W, 3), values in [0, 255].	required
`augmenter`	`TextureAugmenter \| ObjectAugmenter`	An instantiated augmenter (TextureAugmenter or ObjectAugmenter).	required

Returns:

Type	Description
`ndarray`	Augmented batch as numpy array of shape (N, H, W, 3), values in [0, 255].

Source code in app/pipelines/preprocessing/augmentation.py

def augment_batch(
    images: np.ndarray,
    augmenter: TextureAugmenter | ObjectAugmenter,
) -> np.ndarray:
    """Apply augmentation to a batch of numpy images.

    Args:
        images: Batch of images as a numpy array of shape (N, H, W, 3), values in [0, 255].
        augmenter: An instantiated augmenter (TextureAugmenter or ObjectAugmenter).

    Returns:
        Augmented batch as numpy array of shape (N, H, W, 3), values in [0, 255].
    """
    augmented: list[Any] = []
    for img_array in images:
        pil_img = Image.fromarray(img_array.astype(np.uint8), mode="RGB")
        aug_img = augmenter(pil_img)
        augmented.append(np.array(aug_img, dtype=np.uint8))
    return np.stack(augmented, axis=0)

`get_augmenter(category: str) -> TextureAugmenter | ObjectAugmenter` ¶

Factory that returns the correct augmenter for a given MVTec category.

This function automatically selects the appropriate augmentation strategy: - Heavy spatial augmentation (TextureAugmenter) for texture categories. - Light photometric augmentation (ObjectAugmenter) for object categories. - Falls back to ObjectAugmenter (conservative) for unknown categories.

Parameters:

Name	Type	Description	Default
`category`	`str`	MVTec AD category name (e.g., 'wood', 'bottle', 'screw').	required

Returns:

Type	Description
`TextureAugmenter \| ObjectAugmenter`	TextureAugmenter if the category is a texture, ObjectAugmenter otherwise.

Source code in app/pipelines/preprocessing/augmentation.py

def get_augmenter(category: str) -> TextureAugmenter | ObjectAugmenter:
    """Factory that returns the correct augmenter for a given MVTec category.

    This function automatically selects the appropriate augmentation strategy:
    - Heavy spatial augmentation (``TextureAugmenter``) for texture categories.
    - Light photometric augmentation (``ObjectAugmenter``) for object categories.
    - Falls back to ``ObjectAugmenter`` (conservative) for unknown categories.

    Args:
        category: MVTec AD category name (e.g., 'wood', 'bottle', 'screw').

    Returns:
        TextureAugmenter if the category is a texture, ObjectAugmenter otherwise.
    """
    category_lower = category.lower().strip()
    if category_lower in TEXTURE_CATEGORIES:
        logger.info("Category '%s' is a texture → using TextureAugmenter (heavy spatial augmentation).", category)
        return TextureAugmenter()
    logger.info("Category '%s' is an object → using ObjectAugmenter (light photometric augmentation).", category)
    return ObjectAugmenter()

`app.pipelines.preprocessing.segmentation` ¶

Foreground extraction and background replacement for the Keras CAE pipeline.

Why Do We Need Foreground Extraction?¶

Industrial components photographed against a background introduce a fundamental problem: the autoencoder wastes representational capacity learning the background (conveyor belt, mounting jig, inspection stage). This background is not the object under inspection.

Even worse, subtle background variations (dust, lighting reflections, shadow changes) can drive up the reconstruction error and produce false positive anomaly detections.

The Two-Step Solution: Segmentation + Background Replacement (BGRP-G)¶

This module implements the BGRP-G (Background Replacement to Grey/Black) strategy:

Segment the foreground: Use classical computer vision to find the component pixels.
Zero-fill the background: Replace all background pixels with solid black (0, 0, 0).

Why Black (zero) as the Replacement Colour? Black = (0, 0, 0) is the most "out-of-distribution" value for typical industrial inspection images, which tend to be brighter and coloured. The autoencoder, trained only on images with black backgrounds, will learn to perfectly reconstruct black background regions with near-zero error. This means the background contributes nothing to the anomaly score - which is exactly what we want.

Important: We must keep colour information in the foreground intact, since colour
defects (e.g., surface discolouration) are valid anomaly types in MVTec.

Classical CV Approach: Otsu + Adaptive Canny¶

Rather than using SAM (Segment Anything Model, which requires ~2.5 GB model weights), we use a fast, dependency-free classical pipeline:

Otsu's Thresholding: A global binarization method that automatically finds the optimal greyscale threshold to separate foreground from background. It maximises the inter-class variance between foreground and background pixel distributions.
Adaptive Canny Edge Detection: Canny finds sharp pixel intensity transitions (edges). The "adaptive" variant sets high/low thresholds automatically from the image's median pixel intensity, making it robust across varying illumination.
Morphological Closing: Fills small holes in the combined binary mask (gaps between edges and the Otsu region) by dilating then eroding with a kernel.
Largest Connected Component: Selects only the single largest foreground blob, discarding small spurious fragments from dust or image noise.

Module Contents¶

OtsuCannySegmentor: Full foreground extraction pipeline.
extract_largest_component: Helper to isolate the largest blob in a binary mask.

`OtsuCannySegmentor` ¶

Foreground segmentation combining Otsu thresholding with Adaptive Canny edge detection.

This class provides a fast, reliable foreground extraction pipeline that works well on the standard MVTec AD inspection setup (component on a uniform background).

Pipeline Steps

Convert RGB to greyscale for efficient threshold computation.
Apply Otsu's global threshold to create a coarse binary foreground mask.
Compute adaptive Canny edge map using the image's median intensity as the threshold anchor.
Combine (OR) the Otsu mask and Canny edges into one binary map.
Apply morphological closing to fill gaps between adjacent edges.
Extract the single largest connected component to remove noise artefacts.
Replace all background pixels (mask = 0) in the original RGB image with black.

Attributes:

Name	Type	Description
`morph_kernel_size`		Side length (pixels) of the square structuring element used for morphological closing. Larger = fills bigger gaps.
`canny_sigma`		Scaling factor applied to the median pixel intensity to derive the Canny low and high thresholds. Higher = fewer, stronger edges detected.

Source code in app/pipelines/preprocessing/segmentation.py

class OtsuCannySegmentor:
    """Foreground segmentation combining Otsu thresholding with Adaptive Canny edge detection.

    This class provides a fast, reliable foreground extraction pipeline that works well
    on the standard MVTec AD inspection setup (component on a uniform background).

    Pipeline Steps:
        1. Convert RGB to greyscale for efficient threshold computation.
        2. Apply Otsu's global threshold to create a coarse binary foreground mask.
        3. Compute adaptive Canny edge map using the image's median intensity as the
           threshold anchor.
        4. Combine (OR) the Otsu mask and Canny edges into one binary map.
        5. Apply morphological closing to fill gaps between adjacent edges.
        6. Extract the single largest connected component to remove noise artefacts.
        7. Replace all background pixels (mask = 0) in the original RGB image with black.

    Attributes:
        morph_kernel_size: Side length (pixels) of the square structuring element used
            for morphological closing. Larger = fills bigger gaps.
        canny_sigma: Scaling factor applied to the median pixel intensity to derive
            the Canny low and high thresholds. Higher = fewer, stronger edges detected.
    """

    def __init__(self, morph_kernel_size: int = 5, canny_sigma: float = 0.33) -> None:
        """Initialise the segmentor with morphological and edge detection parameters.

        Args:
            morph_kernel_size: Side length of the square kernel for morphological closing.
                5 pixels works well for most MVTec categories.
            canny_sigma: Controls the spread of Canny threshold bounds around the
                image median. Larger values = more conservative edge detection.
        """
        self.morph_kernel_size = morph_kernel_size
        self.canny_sigma = canny_sigma
        self._kernel = cv2.getStructuringElement(
            cv2.MORPH_RECT,
            (morph_kernel_size, morph_kernel_size),
        )

    def compute_mask(self, image_rgb: np.ndarray) -> np.ndarray:
        """Compute a binary foreground mask for the input RGB image.

        Args:
            image_rgb: Input image as a numpy array of shape (H, W, 3), dtype uint8, RGB order.

        Returns:
            Binary mask as a 2D numpy array (H, W), dtype uint8, values 0 or 255.
            255 = foreground (component), 0 = background.
        """
        # Step 1: Convert RGB to greyscale
        # Greyscale = 0.299R + 0.587G + 0.114B (standard luminance formula)
        grey = cv2.cvtColor(image_rgb, cv2.COLOR_RGB2GRAY)

        # Step 2: Otsu global thresholding
        # Otsu automatically finds the threshold that maximises inter-class variance.
        # THRESH_BINARY_INV inverts so the foreground (usually darker component on
        # lighter background) becomes 255.
        _, otsu_mask = cv2.threshold(grey, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)

        # Step 3: Adaptive Canny edge detection
        # Derive thresholds from the image's median pixel intensity:
        # - low  threshold = (1 - sigma) * median
        # - high threshold = (1 + sigma) * median
        median_val = float(np.median(grey))
        low_thresh = max(0.0, (1.0 - self.canny_sigma) * median_val)
        high_thresh = min(255.0, (1.0 + self.canny_sigma) * median_val)
        canny_edges = cv2.Canny(grey, low_thresh, high_thresh)

        # Step 4: Combine Otsu mask and Canny edges (logical OR)
        combined = cv2.bitwise_or(otsu_mask, canny_edges)

        # Step 5: Morphological closing to fill small internal holes
        # Closing = dilate then erode → expands foreground then shrinks back,
        # but keeps filled any gaps smaller than the kernel.
        closed = cv2.morphologyEx(combined, cv2.MORPH_CLOSE, self._kernel)

        # Step 6: Keep only the largest connected foreground blob
        largest = extract_largest_component(closed)

        return largest

    def apply(self, image_rgb: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
        """Extract foreground and replace background with black (BGRP-G strategy).

        The BGRP-G strategy (Background Replacement with a Guaranteed Out-of-Distribution
        colour) zeroes out all background pixels. This forces the autoencoder to learn only
        from the component surface, eliminating background as a source of false anomalies.

        Args:
            image_rgb: Input RGB image as numpy array (H, W, 3), dtype uint8.

        Returns:
            Tuple of:
                - masked_image: RGB image with background pixels zeroed, shape (H, W, 3).
                - foreground_mask: Binary foreground mask, shape (H, W), values 0 or 255.
        """
        foreground_mask = self.compute_mask(image_rgb)

        # Expand mask to 3 channels so it can multiply with RGB image
        mask_3ch = np.stack([foreground_mask] * 3, axis=-1)  # (H, W, 3)

        # Apply mask: foreground stays, background becomes 0 (black)
        masked_image = (image_rgb * (mask_3ch > 0)).astype(np.uint8)

        logger.debug(
            "Segmentation complete. Foreground pixels: %d / %d (%.1f%%)",
            int(np.sum(foreground_mask > 0)),
            foreground_mask.size,
            100.0 * np.sum(foreground_mask > 0) / foreground_mask.size,
        )

        return masked_image, foreground_mask

`init(morph_kernel_size: int = 5, canny_sigma: float = 0.33) -> None` ¶

Initialise the segmentor with morphological and edge detection parameters.

Parameters:

Name	Type	Description	Default
`morph_kernel_size`	`int`	Side length of the square kernel for morphological closing. 5 pixels works well for most MVTec categories.	`5`
`canny_sigma`	`float`	Controls the spread of Canny threshold bounds around the image median. Larger values = more conservative edge detection.	`0.33`

Source code in app/pipelines/preprocessing/segmentation.py

def __init__(self, morph_kernel_size: int = 5, canny_sigma: float = 0.33) -> None:
    """Initialise the segmentor with morphological and edge detection parameters.

    Args:
        morph_kernel_size: Side length of the square kernel for morphological closing.
            5 pixels works well for most MVTec categories.
        canny_sigma: Controls the spread of Canny threshold bounds around the
            image median. Larger values = more conservative edge detection.
    """
    self.morph_kernel_size = morph_kernel_size
    self.canny_sigma = canny_sigma
    self._kernel = cv2.getStructuringElement(
        cv2.MORPH_RECT,
        (morph_kernel_size, morph_kernel_size),
    )

`apply(image_rgb: np.ndarray) -> tuple[np.ndarray, np.ndarray]` ¶

Extract foreground and replace background with black (BGRP-G strategy).

The BGRP-G strategy (Background Replacement with a Guaranteed Out-of-Distribution colour) zeroes out all background pixels. This forces the autoencoder to learn only from the component surface, eliminating background as a source of false anomalies.

Parameters:

Name	Type	Description	Default
`image_rgb`	`ndarray`	Input RGB image as numpy array (H, W, 3), dtype uint8.	required

Returns:

Type	Description
`tuple[ndarray, ndarray]`	Tuple of: - masked_image: RGB image with background pixels zeroed, shape (H, W, 3). - foreground_mask: Binary foreground mask, shape (H, W), values 0 or 255.

Source code in app/pipelines/preprocessing/segmentation.py

def apply(self, image_rgb: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
    """Extract foreground and replace background with black (BGRP-G strategy).

    The BGRP-G strategy (Background Replacement with a Guaranteed Out-of-Distribution
    colour) zeroes out all background pixels. This forces the autoencoder to learn only
    from the component surface, eliminating background as a source of false anomalies.

    Args:
        image_rgb: Input RGB image as numpy array (H, W, 3), dtype uint8.

    Returns:
        Tuple of:
            - masked_image: RGB image with background pixels zeroed, shape (H, W, 3).
            - foreground_mask: Binary foreground mask, shape (H, W), values 0 or 255.
    """
    foreground_mask = self.compute_mask(image_rgb)

    # Expand mask to 3 channels so it can multiply with RGB image
    mask_3ch = np.stack([foreground_mask] * 3, axis=-1)  # (H, W, 3)

    # Apply mask: foreground stays, background becomes 0 (black)
    masked_image = (image_rgb * (mask_3ch > 0)).astype(np.uint8)

    logger.debug(
        "Segmentation complete. Foreground pixels: %d / %d (%.1f%%)",
        int(np.sum(foreground_mask > 0)),
        foreground_mask.size,
        100.0 * np.sum(foreground_mask > 0) / foreground_mask.size,
    )

    return masked_image, foreground_mask

`compute_mask(image_rgb: np.ndarray) -> np.ndarray` ¶

Compute a binary foreground mask for the input RGB image.

Parameters:

Name	Type	Description	Default
`image_rgb`	`ndarray`	Input image as a numpy array of shape (H, W, 3), dtype uint8, RGB order.	required

Returns:

Type	Description
`ndarray`	Binary mask as a 2D numpy array (H, W), dtype uint8, values 0 or 255.
`ndarray`	255 = foreground (component), 0 = background.

Source code in app/pipelines/preprocessing/segmentation.py

def compute_mask(self, image_rgb: np.ndarray) -> np.ndarray:
    """Compute a binary foreground mask for the input RGB image.

    Args:
        image_rgb: Input image as a numpy array of shape (H, W, 3), dtype uint8, RGB order.

    Returns:
        Binary mask as a 2D numpy array (H, W), dtype uint8, values 0 or 255.
        255 = foreground (component), 0 = background.
    """
    # Step 1: Convert RGB to greyscale
    # Greyscale = 0.299R + 0.587G + 0.114B (standard luminance formula)
    grey = cv2.cvtColor(image_rgb, cv2.COLOR_RGB2GRAY)

    # Step 2: Otsu global thresholding
    # Otsu automatically finds the threshold that maximises inter-class variance.
    # THRESH_BINARY_INV inverts so the foreground (usually darker component on
    # lighter background) becomes 255.
    _, otsu_mask = cv2.threshold(grey, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)

    # Step 3: Adaptive Canny edge detection
    # Derive thresholds from the image's median pixel intensity:
    # - low  threshold = (1 - sigma) * median
    # - high threshold = (1 + sigma) * median
    median_val = float(np.median(grey))
    low_thresh = max(0.0, (1.0 - self.canny_sigma) * median_val)
    high_thresh = min(255.0, (1.0 + self.canny_sigma) * median_val)
    canny_edges = cv2.Canny(grey, low_thresh, high_thresh)

    # Step 4: Combine Otsu mask and Canny edges (logical OR)
    combined = cv2.bitwise_or(otsu_mask, canny_edges)

    # Step 5: Morphological closing to fill small internal holes
    # Closing = dilate then erode → expands foreground then shrinks back,
    # but keeps filled any gaps smaller than the kernel.
    closed = cv2.morphologyEx(combined, cv2.MORPH_CLOSE, self._kernel)

    # Step 6: Keep only the largest connected foreground blob
    largest = extract_largest_component(closed)

    return largest

`extract_largest_component(binary_mask: np.ndarray) -> np.ndarray` ¶

Extract only the largest connected foreground region from a binary mask.

After Otsu + Canny segmentation, the mask may contain multiple disconnected blobs (e.g., the main component plus dust particles or image artefacts). This function keeps only the largest blob, which is almost always the actual component.

How it works

Label all connected components in the binary mask.
Count pixels in each component.
Return a mask with only the largest component filled.

Parameters:

Name	Type	Description	Default
`binary_mask`	`ndarray`	2D binary numpy array (dtype uint8), 255 = foreground, 0 = background.	required

Returns:

Type	Description
`ndarray`	Cleaned 2D binary mask with only the largest connected component kept (uint8, 0/255).

Source code in app/pipelines/preprocessing/segmentation.py

def extract_largest_component(binary_mask: np.ndarray) -> np.ndarray:
    """Extract only the largest connected foreground region from a binary mask.

    After Otsu + Canny segmentation, the mask may contain multiple disconnected blobs
    (e.g., the main component plus dust particles or image artefacts). This function
    keeps only the largest blob, which is almost always the actual component.

    How it works:
        1. Label all connected components in the binary mask.
        2. Count pixels in each component.
        3. Return a mask with only the largest component filled.

    Args:
        binary_mask: 2D binary numpy array (dtype uint8), 255 = foreground, 0 = background.

    Returns:
        Cleaned 2D binary mask with only the largest connected component kept (uint8, 0/255).
    """
    # cv2.connectedComponentsWithStats returns:
    # - num_labels: total number of components found (including background = label 0)
    # - labels: 2D array where each pixel has its component label
    # - stats: per-component statistics (bounding box, area)
    # - centroids: per-component centroid (x, y)
    num_labels, labels, stats, _ = cv2.connectedComponentsWithStats(binary_mask, connectivity=8)

    if num_labels <= 1:
        # Only background found (empty mask), return as-is
        return binary_mask

    # stats[i, cv2.CC_STAT_AREA] gives pixel count of component i
    # Label 0 is the background - skip it by starting from label 1
    component_areas = stats[1:, cv2.CC_STAT_AREA]
    largest_label = int(np.argmax(component_areas)) + 1  # +1 to re-align with labels array

    # Build new mask with only the largest component
    largest_mask = np.zeros_like(binary_mask)
    largest_mask[labels == largest_label] = 255

    return largest_mask

Evaluation Subpackage¶

`app.pipelines.evaluation.metrics` ¶

Precision-recall metric calculation and persistence functions.

`canonicalize_pixel_inputs(anomaly_maps: list[np.ndarray] | np.ndarray, masks: list[np.ndarray | None] | np.ndarray, image_labels: np.ndarray | list[int], size: tuple[int, int] = CANONICAL_MAP_SIZE) -> tuple[np.ndarray, np.ndarray]` ¶

Validate and resize full pixel maps and masks to the shared canonical resolution.

Parameters:

Name	Type	Description	Default
`anomaly_maps`	`list[ndarray] \| ndarray`	Sequence of continuous 2D anomaly heatmaps.	required
`masks`	`list[ndarray \| None] \| ndarray`	Sequence of ground-truth binary masks (or None for normal images).	required
`image_labels`	`ndarray \| list[int]`	Binary image-level defect labels (0 or 1).	required
`size`	`tuple[int, int]`	Target canonical resolution tuple (height, width). Defaults to 256x256.	`CANONICAL_MAP_SIZE`

Returns:

Type	Description
`tuple[ndarray, ndarray]`	A tuple of stacked (canonical_maps, canonical_masks) NumPy arrays.

Raises:

Type	Description
`ValueError`	If inputs are empty, have mismatched lengths, or contain invalid labels.

Source code in app/pipelines/evaluation/metrics.py

def canonicalize_pixel_inputs(
    anomaly_maps: list[np.ndarray] | np.ndarray,
    masks: list[np.ndarray | None] | np.ndarray,
    image_labels: np.ndarray | list[int],
    size: tuple[int, int] = CANONICAL_MAP_SIZE,
) -> tuple[np.ndarray, np.ndarray]:
    """Validate and resize full pixel maps and masks to the shared canonical resolution.

    Args:
        anomaly_maps: Sequence of continuous 2D anomaly heatmaps.
        masks: Sequence of ground-truth binary masks (or None for normal images).
        image_labels: Binary image-level defect labels (0 or 1).
        size: Target canonical resolution tuple (height, width). Defaults to 256x256.

    Returns:
        A tuple of stacked (canonical_maps, canonical_masks) NumPy arrays.

    Raises:
        ValueError: If inputs are empty, have mismatched lengths, or contain invalid labels.
    """
    maps_list = list(anomaly_maps)
    masks_list = list(masks)
    labels = np.asarray(image_labels, dtype=np.uint8).reshape(-1)
    if not maps_list:
        raise ValueError("Pixel evaluation requires at least one anomaly map")
    if len(maps_list) != len(masks_list) or len(maps_list) != len(labels):
        raise ValueError("Anomaly maps, masks, and image labels must have equal counts")
    if not set(np.unique(labels)).issubset({0, 1}):
        raise ValueError("Image labels must be binary")

    target_height, target_width = size
    canonical_maps = [
        _canonicalize_single_map(raw_map, target_width, target_height, idx) for idx, raw_map in enumerate(maps_list)
    ]
    canonical_masks = [
        _canonicalize_single_mask(raw_mask, int(label), target_width, target_height, idx)
        for idx, (raw_mask, label) in enumerate(zip(masks_list, labels, strict=True))
    ]

    return np.stack(canonical_maps), np.stack(canonical_masks)

`compute_and_save_pr_metrics(y_true: Any, y_score: Any, output_path: str | Path, level: str = 'pixel', aupimo: float | None = None, fpr_bounds: tuple[float, float] | None = None) -> Path` ¶

Compute PR metrics and save them with an optional genuine AUPIMO score.

Parameters:

Name	Type	Description	Default
`y_true`	`Any`	1D array of ground truth binary labels (0 or 1).	required
`y_score`	`Any`	1D array of predicted anomaly scores.	required
`output_path`	`str \| Path`	Destination .npz file path.	required
`level`	`str`	Evaluation level ('pixel' for localization, 'image' for classification).	`'pixel'`
`aupimo`	`float \| None`	Genuine full-map AUPIMO score computed separately from 2D maps.	`None`
`fpr_bounds`	`tuple[float, float] \| None`	FPR integration bounds used for AUPIMO.	`None`

Returns:

Type	Description
`Path`	The saved Path object.

Source code in app/pipelines/evaluation/metrics.py

def compute_and_save_pr_metrics(
    y_true: Any,
    y_score: Any,
    output_path: str | Path,
    level: str = "pixel",
    aupimo: float | None = None,
    fpr_bounds: tuple[float, float] | None = None,
) -> Path:
    """Compute PR metrics and save them with an optional genuine AUPIMO score.

    Args:
        y_true: 1D array of ground truth binary labels (0 or 1).
        y_score: 1D array of predicted anomaly scores.
        output_path: Destination .npz file path.
        level: Evaluation level ('pixel' for localization, 'image' for classification).
        aupimo: Genuine full-map AUPIMO score computed separately from 2D maps.
        fpr_bounds: FPR integration bounds used for AUPIMO.

    Returns:
        The saved Path object.
    """
    y_true_arr = np.asarray(y_true)
    y_score_arr = np.asarray(y_score)

    precision, recall, thresholds = precision_recall_curve(y_true_arr, y_score_arr)

    return save_evaluation_metrics(
        output_path,
        precision,
        recall,
        thresholds,
        aupimo=aupimo,
        fpr_bounds=fpr_bounds,
        level=level,
    )

`compute_aupimo(anomaly_maps: list[np.ndarray], gt_masks: list[np.ndarray | None], fpr_bounds: tuple[float, float] = (1e-05, 0.0001)) -> float` ¶

Compute pixel-level AUPIMO using anomalib's implementation.

AUPIMO (Area Under Per-Image Overlap) integrates per-image pixel overlap between predicted anomaly maps and ground truth defect masks. It does so only over an extremely narrow and industrially realistic FPR range (default: 10⁻⁵ to 10⁻⁴).

Parameters:

Name	Type	Description	Default
`anomaly_maps`	`list[ndarray]`	List of 2D pixel anomaly score maps, one per test image. Each map has shape (H, W) with float values >= 0 (higher = more anomalous).	required
`gt_masks`	`list[ndarray \| None]`	List of 2D ground truth binary masks, one per test image. Each mask has shape (H, W) with values 0 (normal) or 1 (defect). Use None for images with no ground truth mask (normal images).	required
`fpr_bounds`	`tuple[float, float]`	Tuple (lower_fpr, upper_fpr) defining the integration interval. Default (1e-5, 1e-4) matches the MVTec AD benchmark standard.	`(1e-05, 0.0001)`

Returns:

Type	Description
`float`	AUPIMO score in [0, 1]. Higher is better.

Raises:

Type	Description
`ImportError`	If anomalib or torch is unavailable.
`ValueError`	If the maps or masks cannot define AUPIMO at the requested bounds.
`RuntimeError`	If anomalib cannot compute the metric at the requested bounds.

Source code in app/pipelines/evaluation/metrics.py

def compute_aupimo(
    anomaly_maps: list[np.ndarray],
    gt_masks: list[np.ndarray | None],
    fpr_bounds: tuple[float, float] = (1e-5, 1e-4),
) -> float:
    """Compute pixel-level AUPIMO using anomalib's implementation.

    AUPIMO (Area Under Per-Image Overlap) integrates per-image pixel overlap between
    predicted anomaly maps and ground truth defect masks. It does so only over an
    extremely narrow and industrially realistic FPR range (default: 10⁻⁵ to 10⁻⁴).

    Args:
        anomaly_maps: List of 2D pixel anomaly score maps, one per test image.
            Each map has shape (H, W) with float values >= 0 (higher = more anomalous).
        gt_masks: List of 2D ground truth binary masks, one per test image.
            Each mask has shape (H, W) with values 0 (normal) or 1 (defect).
            Use None for images with no ground truth mask (normal images).
        fpr_bounds: Tuple (lower_fpr, upper_fpr) defining the integration interval.
            Default (1e-5, 1e-4) matches the MVTec AD benchmark standard.

    Returns:
        AUPIMO score in [0, 1]. Higher is better.

    Raises:
        ImportError: If anomalib or torch is unavailable.
        ValueError: If the maps or masks cannot define AUPIMO at the requested bounds.
        RuntimeError: If anomalib cannot compute the metric at the requested bounds.
    """
    try:
        import torch
        from anomalib.data import ImageBatch
        from anomalib.metrics import AUPIMO
    except ImportError as exc:
        raise ImportError("anomalib and torch are required to compute AUPIMO") from exc

    # Format masks: normal images have zeros mask, defective have binary mask
    h, w = anomaly_maps[0].shape
    all_masks: list[np.ndarray] = []
    has_anomaly = False
    for mask in gt_masks:
        if mask is not None and np.any(mask > 0):
            all_masks.append(mask.astype(np.uint8))
            has_anomaly = True
        else:
            all_masks.append(np.zeros((h, w), dtype=np.uint8))

    if not has_anomaly:
        raise ValueError("AUPIMO requires at least one anomalous image with a ground-truth mask")

    pred_tensor = torch.tensor(np.stack(anomaly_maps), dtype=torch.float32)
    gt_tensor = torch.tensor(np.stack(all_masks), dtype=torch.bool)
    dummy_img = torch.zeros(len(anomaly_maps), 3, h, w, dtype=torch.float32)

    batch = ImageBatch(image=dummy_img, anomaly_map=pred_tensor, gt_mask=gt_tensor)

    aupimo_metric = AUPIMO(num_thresholds=50_000, fpr_bounds=fpr_bounds)
    aupimo_metric.update(batch)
    result = aupimo_metric.compute()
    if hasattr(result, "aupimo_scores"):
        score = float(result.aupimo_scores.nanmean().item())
    elif isinstance(result, tuple) and len(result) > 1:
        score = float(result[1].nanmean().item())
    elif isinstance(result, dict):
        score = float(next(iter(result.values())))
    else:
        score = float(result)

    if np.isnan(score):
        raise RuntimeError(
            f"AUPIMO computation returned NaN for fpr_bounds={fpr_bounds}. "
            "Verify that anomaly maps contain sufficiently diverse continuous scores."
        )

    logger.info("Pixel-Level AUPIMO: %.4f (bounds: %s)", score, fpr_bounds)
    return score

`compute_image_auroc(scores: np.ndarray, binary_labels: np.ndarray) -> float` ¶

Compute image-level Area Under the ROC Curve (AUROC).

Parameters:

Name	Type	Description	Default
`scores`	`ndarray`	1D array of image-level anomaly scores, shape (N,). Higher = more anomalous.	required
`binary_labels`	`ndarray`	1D binary array, shape (N,). 0 = normal, 1 = anomalous.	required

Returns:

Type	Description
`float`	AUROC value in [0, 1]. 1.0 = perfect; 0.5 = random; 0.0 = perfectly inverted.

Raises:

Type	Description
`ValueError`	If fewer than 2 distinct classes are present in binary_labels.

Source code in app/pipelines/evaluation/metrics.py

def compute_image_auroc(scores: np.ndarray, binary_labels: np.ndarray) -> float:
    """Compute image-level Area Under the ROC Curve (AUROC).

    Args:
        scores: 1D array of image-level anomaly scores, shape (N,). Higher = more anomalous.
        binary_labels: 1D binary array, shape (N,). 0 = normal, 1 = anomalous.

    Returns:
        AUROC value in [0, 1]. 1.0 = perfect; 0.5 = random; 0.0 = perfectly inverted.

    Raises:
        ValueError: If fewer than 2 distinct classes are present in binary_labels.
    """
    if len(np.unique(binary_labels)) < 2:
        raise ValueError("Image AUROC requires both normal and anomalous labels")

    auroc: float = float(roc_auc_score(binary_labels, scores))
    logger.info("Image-Level AUROC: %.4f", auroc)
    return auroc

`compute_image_confusion_metrics(labels: Any, scores: Any, threshold: float) -> dict[str, float | int]` ¶

Calculate image confusion counts and derived metrics at a frozen threshold.

Source code in app/pipelines/evaluation/metrics.py

def compute_image_confusion_metrics(labels: Any, scores: Any, threshold: float) -> dict[str, float | int]:
    """Calculate image confusion counts and derived metrics at a frozen threshold."""
    y_true = np.asarray(labels, dtype=np.uint8).reshape(-1)
    y_score = np.asarray(scores, dtype=np.float64).reshape(-1)
    if len(y_true) == 0 or len(y_true) != len(y_score):
        raise ValueError("Image labels and scores must be non-empty and have equal counts")
    if not set(np.unique(y_true)).issubset({0, 1}):
        raise ValueError("Image labels must be binary")
    predictions = (y_score > threshold).astype(np.uint8)
    true_negatives, false_positives, false_negatives, true_positives = confusion_matrix(
        y_true, predictions, labels=[0, 1]
    ).ravel()
    precision = true_positives / (true_positives + false_positives) if true_positives + false_positives else 0.0
    recall = true_positives / (true_positives + false_negatives) if true_positives + false_negatives else 0.0
    f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
    return {
        "true_positives": int(true_positives),
        "false_positives": int(false_positives),
        "false_negatives": int(false_negatives),
        "true_negatives": int(true_negatives),
        "precision": float(precision),
        "recall": float(recall),
        "f1_score": float(f1),
    }

`compute_shared_pixel_metrics(anomaly_maps: list[np.ndarray] | np.ndarray, masks: list[np.ndarray | None] | np.ndarray, image_labels: np.ndarray | list[int]) -> tuple[dict[str, Any], np.ndarray, np.ndarray]` ¶

Compute canonical pixel AUROC and genuine full-map AUPIMO.

Source code in app/pipelines/evaluation/metrics.py

def compute_shared_pixel_metrics(
    anomaly_maps: list[np.ndarray] | np.ndarray,
    masks: list[np.ndarray | None] | np.ndarray,
    image_labels: np.ndarray | list[int],
) -> tuple[dict[str, Any], np.ndarray, np.ndarray]:
    """Compute canonical pixel AUROC and genuine full-map AUPIMO."""
    canonical_maps, canonical_masks = canonicalize_pixel_inputs(anomaly_maps, masks, image_labels)
    flat_masks = canonical_masks.reshape(-1)
    if len(np.unique(flat_masks)) != 2:
        raise ValueError("Pixel AUROC requires both normal and anomalous pixels")
    pixel_auroc = float(roc_auc_score(flat_masks, canonical_maps.reshape(-1)))

    # Import lazily to avoid an evaluation-module import cycle and to keep this
    # helper patchable in focused tests.
    from app.pipelines.evaluation.cae_metrics import compute_aupimo

    pixel_aupimo = compute_aupimo(
        [item for item in canonical_maps],
        [item for item in canonical_masks],
        fpr_bounds=AUPIMO_FPR_BOUNDS,
    )
    return (
        {
            "pixel_auroc": pixel_auroc,
            "pixel_aupimo": pixel_aupimo,
            "aupimo_fpr_lower": AUPIMO_FPR_BOUNDS[0],
            "aupimo_fpr_upper": AUPIMO_FPR_BOUNDS[1],
            "aupimo_num_thresholds": AUPIMO_NUM_THRESHOLDS,
            "canonical_height": CANONICAL_MAP_SIZE[0],
            "canonical_width": CANONICAL_MAP_SIZE[1],
            "pixel_metrics_version": PIXEL_METRICS_VERSION,
        },
        canonical_maps,
        canonical_masks,
    )

`fair_metric_evidence() -> dict[str, Any]` ¶

Return the metric and calibration fields required for a fair cache hit.

Source code in app/pipelines/evaluation/metrics.py

def fair_metric_evidence() -> dict[str, Any]:
    """Return the metric and calibration fields required for a fair cache hit."""
    return {
        "canonical_height": CANONICAL_MAP_SIZE[0],
        "canonical_width": CANONICAL_MAP_SIZE[1],
        "aupimo_fpr_bounds": list(AUPIMO_FPR_BOUNDS),
        "aupimo_num_thresholds": AUPIMO_NUM_THRESHOLDS,
        "threshold_source": "normal_validation",
        "pixel_metrics_version": PIXEL_METRICS_VERSION,
    }

`save_evaluation_metrics(output_path: str | Path, precisions: Any, recalls: Any, thresholds: Any, aupimo: float | None = None, fpr_bounds: tuple[float, float] | None = None, level: str = 'pixel') -> Path` ¶

Save precision, recall, and threshold arrays to an .npz file.

Parameters:

Name	Type	Description	Default
`output_path`	`str \| Path`	Target filepath (e.g. 'results/Patchcore/bottle/pixel_metrics.npz').	required
`precisions`	`Any`	Precision values array.	required
`recalls`	`Any`	Recall values array.	required
`thresholds`	`Any`	Binarization thresholds array.	required
`aupimo`	`float \| None`	Genuine full-map AUPIMO score, when available.	`None`
`fpr_bounds`	`tuple[float, float] \| None`	FPR integration bounds used for AUPIMO, when available.	`None`
`level`	`str`	Evaluation level ('pixel' for localization, 'image' for classification).	`'pixel'`

Returns:

Type	Description
`Path`	The saved Path object.

Source code in app/pipelines/evaluation/metrics.py

def save_evaluation_metrics(
    output_path: str | Path,
    precisions: Any,
    recalls: Any,
    thresholds: Any,
    aupimo: float | None = None,
    fpr_bounds: tuple[float, float] | None = None,
    level: str = "pixel",
) -> Path:
    """Save precision, recall, and threshold arrays to an ``.npz`` file.

    Args:
        output_path: Target filepath (e.g. 'results/Patchcore/bottle/pixel_metrics.npz').
        precisions: Precision values array.
        recalls: Recall values array.
        thresholds: Binarization thresholds array.
        aupimo: Genuine full-map AUPIMO score, when available.
        fpr_bounds: FPR integration bounds used for AUPIMO, when available.
        level: Evaluation level ('pixel' for localization, 'image' for classification).

    Returns:
        The saved Path object.
    """
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    values = {
        "precision": precisions,
        "recall": recalls,
        "thresholds": thresholds,
        "level": level,
    }
    if aupimo is not None:
        values["aupimo"] = aupimo
    if fpr_bounds is not None:
        values["aupimo_fpr_bounds"] = np.asarray(fpr_bounds, dtype=np.float64)
    np.savez(path, **values)
    return path

`app.pipelines.evaluation.cae_metrics` ¶

Evaluation metrics and heatmap generation for the Keras CAE pipeline.

Why Evaluation Methodology Matters¶

Getting the right evaluation metric is as important as the model itself. Using the wrong metric can make a terrible detector look great on paper, and vice versa.

Image-Level: AUROC (Area Under ROC Curve)¶

AUROC measures how well the model ranks anomalous images above normal ones across ALL possible thresholds simultaneously. An AUROC of 1.0 means perfect ranking; 0.5 means the model is no better than random guessing.

Advantages over plain accuracy: - Threshold-independent: does not require choosing a specific cut-off. - Handles class imbalance well (MVTec test sets are typically imbalanced).

Pixel-Level: AUPIMO vs. PRO-Score¶

For pixel-level evaluation (localising where the defect is), two metrics exist:

PRO-Score (Per-Region Overlap / AUPRO) Integrates overlap between predicted anomaly maps and ground-truth masks across thresholds up to a fixed FPR on normal images. In practice it over-weights tiny label annotation errors, which are common in real industrial datasets.

AUPIMO (Area Under Per-Image Overlap) ← Used here AUPIMO introduces two critical improvements:

1. **Normal-Only Validation**: Thresholds are calibrated exclusively on images
   with zero defects. This prevents the metric from being "gamed" by correctly
   identifying easy normal regions.

2. **Logarithmic FPR Bounds**: Integration happens only between FPR = 10⁻⁵ and
   FPR = 10⁻⁴. This extremely tight range corresponds to real industrial reject
   rates (maximum 1 false alarm per 10,000-100,000 inspected parts).

The result is a metric that honestly reflects real industrial performance, not
laboratory performance under lenient conditions.

Heatmap Generation¶

Raw pixel error maps need normalisation before visualisation, because: - Absolute error values depend on the model's training quality. - Different images have different baseline error levels.

We use quantile normalisation: clamp to the 1st and 99th percentile of the error distribution, then rescale to [0, 255]. This prevents a few outlier pixels from washing out the rest of the heatmap.

Module Contents¶

compute_image_auroc: Image-level ROC AUC from scores and binary labels.
compute_aupimo: Pixel-level AUPIMO using anomalib's implementation.
generate_heatmap_overlay: Creates an RGB overlay of error on the original image.
evaluate_cae: Full evaluation pipeline returning all metrics.

`evaluate_cae(model: Any, test_images: np.ndarray, test_labels: np.ndarray, gt_masks: list[np.ndarray | None], threshold: float, k_fraction: float = 0.002, output_dir: Path | None = None, reconstructions: np.ndarray | None = None) -> dict[str, Any]` ¶

Run full evaluation of the trained CAE on the test set.

Computes: - Image-level anomaly scores using Top-K pooling. - AUROC across all test images. - Accuracy, precision, recall using the calibrated adaptive threshold. - AUPIMO for pixel-level localisation on images with ground truth masks.

Parameters:

Name	Type	Description	Default
`model`	`Any`	Trained Keras CAE model.	required
`test_images`	`ndarray`	Normalised test images, shape (N, H, W, 3), values in [0, 1].	required
`test_labels`	`ndarray`	Binary labels, shape (N,). 0 = normal, 1 = anomalous.	required
`gt_masks`	`list[ndarray \| None]`	List of ground truth defect masks (or None for normal images).	required
`threshold`	`float`	Decision threshold from `compute_adaptive_threshold`.	required
`k_fraction`	`float`	Top-K pooling fraction for image-level scoring.	`0.002`
`output_dir`	`Path \| None`	Directory to save detailed PR metrics (.npz files) for UI rendering.	`None`
`reconstructions`	`ndarray \| None`	Optional pre-computed full-image reconstructions.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dictionary containing all evaluation results:
`dict[str, Any]`	`"auroc"`: Image-level AUROC.
`dict[str, Any]`	`"aupimo"`: Pixel-level AUPIMO.
`dict[str, Any]`	`"accuracy"`: Classification accuracy at the given threshold.
`dict[str, Any]`	`"precision"`: Classification precision at the given threshold.
`dict[str, Any]`	`"recall"`: Classification recall at the given threshold.
`dict[str, Any]`	`"f1_score"`: Classification F1 score at the given threshold.
`dict[str, Any]`	`"scores"`: Raw image anomaly scores (numpy array).
`dict[str, Any]`	`"error_maps"`: List of 2D pixel error maps.
`dict[str, Any]`	`"threshold"`: The decision threshold used.

Source code in app/pipelines/evaluation/cae_metrics.py

def evaluate_cae(
    model: Any,
    test_images: np.ndarray,
    test_labels: np.ndarray,
    gt_masks: list[np.ndarray | None],
    threshold: float,
    k_fraction: float = 0.002,
    output_dir: Path | None = None,
    reconstructions: np.ndarray | None = None,
) -> dict[str, Any]:
    """Run full evaluation of the trained CAE on the test set.

    Computes:
    - Image-level anomaly scores using Top-K pooling.
    - AUROC across all test images.
    - Accuracy, precision, recall using the calibrated adaptive threshold.
    - AUPIMO for pixel-level localisation on images with ground truth masks.

    Args:
        model: Trained Keras CAE model.
        test_images: Normalised test images, shape (N, H, W, 3), values in [0, 1].
        test_labels: Binary labels, shape (N,). 0 = normal, 1 = anomalous.
        gt_masks: List of ground truth defect masks (or None for normal images).
        threshold: Decision threshold from ``compute_adaptive_threshold``.
        k_fraction: Top-K pooling fraction for image-level scoring.
        output_dir: Directory to save detailed PR metrics (.npz files) for UI rendering.
        reconstructions: Optional pre-computed full-image reconstructions.

    Returns:
        Dictionary containing all evaluation results:
        - ``"auroc"``: Image-level AUROC.
        - ``"aupimo"``: Pixel-level AUPIMO.
        - ``"accuracy"``: Classification accuracy at the given threshold.
        - ``"precision"``: Classification precision at the given threshold.
        - ``"recall"``: Classification recall at the given threshold.
        - ``"f1_score"``: Classification F1 score at the given threshold.
        - ``"scores"``: Raw image anomaly scores (numpy array).
        - ``"error_maps"``: List of 2D pixel error maps.
        - ``"threshold"``: The decision threshold used.
    """
    from sklearn.metrics import accuracy_score

    from app.pipelines.evaluation.metrics import (
        AUPIMO_FPR_BOUNDS,
        compute_and_save_pr_metrics,
        compute_image_confusion_metrics,
        compute_shared_pixel_metrics,
    )
    from app.pipelines.evaluation.scoring import compute_image_scores

    logger.info("Running CAE evaluation on %d test images...", len(test_images))

    scores, error_maps = compute_image_scores(
        model, test_images, k_fraction=k_fraction, reconstructions=reconstructions
    )
    binary_labels = test_labels.astype(int)

    auroc = compute_image_auroc(scores, binary_labels)
    pixel_metrics, canonical_maps, canonical_masks = compute_shared_pixel_metrics(error_maps, gt_masks, binary_labels)
    aupimo = float(pixel_metrics["pixel_aupimo"])

    predictions = (scores > threshold).astype(int)
    acc = float(accuracy_score(binary_labels, predictions))
    confusion = compute_image_confusion_metrics(binary_labels, scores, threshold)
    prec = float(confusion["precision"])
    rec = float(confusion["recall"])
    f1 = float(confusion["f1_score"])

    logger.info(
        "Evaluation complete — AUROC: %.4f | AUPIMO: %.4f | Acc: %.2f%% | Prec: %.2f | Rec: %.2f",
        auroc,
        aupimo,
        acc * 100,
        prec,
        rec,
    )

    pixel_auroc = float(pixel_metrics["pixel_auroc"])
    pixel_f1 = 0.0
    if output_dir:
        compute_and_save_pr_metrics(binary_labels, scores, output_dir / "image_metrics.npz", level="image")

        y_true_pixel = canonical_masks.reshape(-1)
        y_score_pixel = canonical_maps.reshape(-1)

        compute_and_save_pr_metrics(
            y_true_pixel,
            y_score_pixel,
            output_dir / "pixel_metrics.npz",
            level="pixel",
            aupimo=aupimo,
            fpr_bounds=AUPIMO_FPR_BOUNDS,
        )

        from sklearn.metrics import precision_score, recall_score

        pixel_pred = (y_score_pixel > threshold).astype(int)
        pixel_prec = float(precision_score(y_true_pixel, pixel_pred, zero_division=0))
        pixel_rec = float(recall_score(y_true_pixel, pixel_pred, zero_division=0))
        pixel_f1 = 2 * (pixel_prec * pixel_rec) / (pixel_prec + pixel_rec) if (pixel_prec + pixel_rec) > 0 else 0.0

    return {
        "auroc": auroc,
        "aupimo": aupimo,
        "accuracy": acc,
        "precision": prec,
        "recall": rec,
        "f1_score": f1,
        "pixel_auroc": pixel_auroc,
        "pixel_f1": pixel_f1,
        "scores": scores,
        "error_maps": error_maps,
        "threshold": threshold,
        **confusion,
        **pixel_metrics,
    }

`generate_heatmap_overlay(original_image: np.ndarray, error_map: np.ndarray, alpha: float = 0.6) -> np.ndarray` ¶

Generate a colour heatmap overlay of the reconstruction error on the original image.

The error map is normalised using robust quantile clamping (1st-99th percentile) to prevent outlier pixels from dominating the colour scale. The heatmap is then blended with the original image.

Colour scheme: - Blue (cool) → low error → likely normal region. - Red (warm) → high error → likely anomalous region.

Parameters:

Name	Type	Description	Default
`original_image`	`ndarray`	RGB image as numpy array, shape (H, W, 3), values in [0, 255] uint8.	required
`error_map`	`ndarray`	2D pixel error map, shape (H, W), values ≥ 0.	required
`alpha`	`float`	Blend weight for the heatmap overlay (0=original only, 1=heatmap only). Default 0.6 → 60% heatmap, 40% original.	`0.6`

Returns:

Type	Description
`ndarray`	RGB overlay image as numpy array, shape (H, W, 3), uint8.

Source code in app/pipelines/evaluation/cae_metrics.py

def generate_heatmap_overlay(
    original_image: np.ndarray,
    error_map: np.ndarray,
    alpha: float = 0.6,
) -> np.ndarray:
    """Generate a colour heatmap overlay of the reconstruction error on the original image.

    The error map is normalised using robust quantile clamping (1st-99th percentile)
    to prevent outlier pixels from dominating the colour scale. The heatmap is then
    blended with the original image.

    Colour scheme:
    - Blue (cool) → low error → likely normal region.
    - Red (warm) → high error → likely anomalous region.

    Args:
        original_image: RGB image as numpy array, shape (H, W, 3), values in [0, 255] uint8.
        error_map: 2D pixel error map, shape (H, W), values ≥ 0.
        alpha: Blend weight for the heatmap overlay (0=original only, 1=heatmap only).
            Default 0.6 → 60% heatmap, 40% original.

    Returns:
        RGB overlay image as numpy array, shape (H, W, 3), uint8.
    """
    import cv2

    # Quantile-normalise: clamp to 1st-99th percentile range to suppress outliers
    p_low = float(np.percentile(error_map, 1))
    p_high = float(np.percentile(error_map, 99))

    if abs(p_high - p_low) < 1e-8:
        # Flat map (no variation) → return original image unchanged
        return original_image.copy()

    normalised = np.clip((error_map - p_low) / (p_high - p_low), 0.0, 1.0)
    heatmap_uint8 = (normalised * 255).astype(np.uint8)

    # Apply OpenCV's JET colormap (blue=low, red=high error)
    heatmap_bgr = cv2.applyColorMap(heatmap_uint8, cv2.COLORMAP_JET)
    heatmap_rgb = cv2.cvtColor(heatmap_bgr, cv2.COLOR_BGR2RGB)

    # Blend heatmap with original image
    original_float = original_image.astype(np.float32)
    heatmap_float = heatmap_rgb.astype(np.float32)
    blended = (alpha * heatmap_float + (1.0 - alpha) * original_float).clip(0, 255).astype(np.uint8)

    return blended

`app.pipelines.evaluation.scoring` ¶

Anomaly scoring algorithms for the Keras CAE pipeline.

This module implements two key improvements over naive anomaly scoring that make the system significantly more robust for real-world industrial inspection:

Top-K Pooling (replaces Max-Pooling for image-level scoring)
Adaptive Thresholding (replaces a fixed hard-coded cut-off)

Why Image-Level Anomaly Scoring Matters¶

The autoencoder produces a 2D error map (pixel-wise reconstruction error). To decide "is this image anomalous?" we need to collapse this map into a single score.

Max-Pooling

Image score = max(error_map) Problem: A single noisy pixel (camera sensor spike, dust particle, JPEG artefact) can drive the max value very high, producing a false positive on a perfectly good part.

Top-K Pooling

Image score = mean(top K highest pixels) Rationale: Real industrial defects (cracks, scratches, contamination patches) always appear as clusters of elevated error pixels, not isolated spikes. By averaging the K highest values, isolated single-pixel noise is diluted, while genuine defect clusters (which affect many pixels together) still produce reliably high scores.

A commonly effective value is K = 0.2% of total pixels. For a 128x128 image = 16,384 pixels -> K ~= 33 pixels.

Why Adaptive Thresholds Are Essential¶

A fixed threshold (e.g., "score > 0.05 = anomalous") will fail when: - Different cameras / lighting conditions shift the absolute score range. - Different MVTec categories (leather vs. metal) have vastly different texture complexity. - Batch-to-batch variation in normal samples changes the baseline reconstruction quality.

Adaptive approaches calibrate the threshold on the normal training/validation data:

quantile method: threshold = np.percentile(normal_scores, 95) Interpretation: "The model is trained; 95% of normal images score below this value. Anything higher is likely anomalous."

mahalanobis method: Models the normal score distribution as a Gaussian. The threshold is set at mean + n_sigma * std. This is more principled than a percentile and is closer to a proper statistical test (rejecting the null hypothesis that the image is normal).

Module Contents¶

compute_pixel_error_map: Computes per-pixel reconstruction error.
top_k_pooling: Aggregates error map to a single image-level score robustly.
compute_image_scores: Scores an entire dataset using Top-K pooling.
compute_adaptive_threshold: Derives a decision boundary from normal score statistics.

`compute_adaptive_threshold(normal_scores: np.ndarray, method: Literal['quantile', 'mahalanobis'] = 'quantile', quantile: float = 0.95, n_sigma: float = 3.0) -> float` ¶

Compute an adaptive anomaly decision threshold from normal image scores.

Rather than using a hand-tuned fixed threshold, this function calibrates the decision boundary using the statistical distribution of normal image scores.

Parameters:

Name	Type	Description	Default
`normal_scores`	`ndarray`	1D array of anomaly scores computed on known-good (normal) images. These are used as the calibration reference.	required
`method`	`Literal['quantile', 'mahalanobis']`	Threshold derivation method: - `"quantile"`: Set threshold at the given percentile of normal scores. Intuitive and non-parametric. Works well when the score distribution is non-Gaussian or has outliers. - `"mahalanobis"`: Fit a Gaussian to normal scores (mean + std), then set threshold at `mean + n_sigma * std`. More statistically principled. Assumes the normal score distribution is approximately Gaussian.	`'quantile'`
`quantile`	`float`	Percentile to use for the quantile method (0 < quantile < 1). Default 0.95 → 95th percentile of normal scores becomes the threshold.	`0.95`
`n_sigma`	`float`	Number of standard deviations above the mean for Mahalanobis method. Default 3.0 → corresponds to a false positive rate of ≈0.13% under Gaussian.	`3.0`

Returns:

Type	Description
`float`	Threshold float value. Images scoring above this are classified as anomalous.

Raises:

Type	Description
`ValueError`	If `normal_scores` is empty or if `method` is not recognised.

Source code in app/pipelines/evaluation/scoring.py

def compute_adaptive_threshold(
    normal_scores: np.ndarray,
    method: Literal["quantile", "mahalanobis"] = "quantile",
    quantile: float = 0.95,
    n_sigma: float = 3.0,
) -> float:
    """Compute an adaptive anomaly decision threshold from normal image scores.

    Rather than using a hand-tuned fixed threshold, this function calibrates the
    decision boundary using the statistical distribution of normal image scores.

    Args:
        normal_scores: 1D array of anomaly scores computed on known-good (normal) images.
            These are used as the calibration reference.
        method: Threshold derivation method:
            - ``"quantile"``: Set threshold at the given percentile of normal scores.
              Intuitive and non-parametric. Works well when the score distribution is
              non-Gaussian or has outliers.
            - ``"mahalanobis"``: Fit a Gaussian to normal scores (mean + std), then set
              threshold at ``mean + n_sigma * std``. More statistically principled.
              Assumes the normal score distribution is approximately Gaussian.
        quantile: Percentile to use for the quantile method (0 < quantile < 1).
            Default 0.95 → 95th percentile of normal scores becomes the threshold.
        n_sigma: Number of standard deviations above the mean for Mahalanobis method.
            Default 3.0 → corresponds to a false positive rate of ≈0.13% under Gaussian.

    Returns:
        Threshold float value. Images scoring above this are classified as anomalous.

    Raises:
        ValueError: If ``normal_scores`` is empty or if ``method`` is not recognised.
    """
    if len(normal_scores) == 0:
        raise ValueError("normal_scores must not be empty for threshold calibration.")

    if method == "quantile":
        threshold = float(np.percentile(normal_scores, quantile * 100))
        logger.info("Quantile threshold (%.0f%%): %.6f", quantile * 100, threshold)

    elif method == "mahalanobis":
        mu = float(np.mean(normal_scores))
        sigma = float(np.std(normal_scores))
        if sigma < 1e-8:
            logger.warning("Normal scores have near-zero variance. Using mean as threshold.")
            threshold = mu
        else:
            threshold = mu + n_sigma * sigma
        logger.info(
            "Mahalanobis threshold (mu + %.1f*sigma): %.6f  (mu=%.6f, sigma=%.6f)",
            n_sigma,
            threshold,
            mu,
            sigma,
        )

    else:
        raise ValueError(f"Unknown threshold method '{method}'. Choose 'quantile' or 'mahalanobis'.")

    return threshold

`compute_image_scores(model: Any, images: np.ndarray, k_fraction: float = 0.002, reconstructions: np.ndarray | None = None) -> tuple[np.ndarray, list[np.ndarray]]` ¶

Compute image-level anomaly scores and pixel error maps for a dataset.

This function runs the trained CAE on every test image, computes the per-pixel error map, and aggregates to an image-level score using Top-K pooling.

Parameters:

Name	Type	Description	Default
`model`	`Any`	Trained Keras CAE model. Must have a `predict` method. Can be None if reconstructions are provided.	required
`images`	`ndarray`	Array of normalised test images, shape (N, H, W, 3), values in [0, 1].	required
`k_fraction`	`float`	Top-K pooling fraction. See `top_k_pooling` for details.	`0.002`
`reconstructions`	`ndarray \| None`	Optional pre-computed reconstructions. If None, uses model.predict.	`None`

Returns:

Type	Description
`tuple[ndarray, list[ndarray]]`	Tuple of: - scores: 1D numpy array of image-level anomaly scores, shape (N,). - error_maps: List of N 2D error maps, each shape (H, W).

Source code in app/pipelines/evaluation/scoring.py

def compute_image_scores(
    model: Any,
    images: np.ndarray,
    k_fraction: float = 0.002,
    reconstructions: np.ndarray | None = None,
) -> tuple[np.ndarray, list[np.ndarray]]:
    """Compute image-level anomaly scores and pixel error maps for a dataset.

    This function runs the trained CAE on every test image, computes the per-pixel
    error map, and aggregates to an image-level score using Top-K pooling.

    Args:
        model: Trained Keras CAE model. Must have a ``predict`` method. Can be None if reconstructions are provided.
        images: Array of normalised test images, shape (N, H, W, 3), values in [0, 1].
        k_fraction: Top-K pooling fraction. See ``top_k_pooling`` for details.
        reconstructions: Optional pre-computed reconstructions. If None, uses model.predict.

    Returns:
        Tuple of:
            - scores: 1D numpy array of image-level anomaly scores, shape (N,).
            - error_maps: List of N 2D error maps, each shape (H, W).
    """
    # Run all images through the CAE in a single batched predict call if not provided
    if reconstructions is None:
        reconstructions = model.predict(images, verbose=0)

    scores: list[float] = []
    error_maps: list[np.ndarray] = []

    for orig, recon in zip(images, reconstructions, strict=True):
        emap = compute_pixel_error_map(orig, recon)
        score = top_k_pooling(emap, k_fraction=k_fraction)
        error_maps.append(emap)
        scores.append(score)

    return np.array(scores, dtype=np.float32), error_maps

`compute_pixel_error_map(original: np.ndarray, reconstruction: np.ndarray, alpha: float = 0.84, sigma: float = 2.0) -> np.ndarray` ¶

Compute the per-pixel absolute reconstruction error map.

The error map is a weighted blend of Structural Similarity (SSIM) error and channel-wise Mean Absolute Error (MAE), aligned with the training loss. A Gaussian blur is applied to smooth noise and cluster anomaly predictions.

Parameters:

Name	Type	Description	Default
`original`	`ndarray`	Original normalised image, shape (H, W, 3), values in [0, 1].	required
`reconstruction`	`ndarray`	Reconstructed image from the CAE, same shape as original.	required
`alpha`	`float`	Weight for the SSIM component (default 0.84, matches loss).	`0.84`
`sigma`	`float`	Standard deviation for Gaussian kernel (default 2.0).	`2.0`

Returns:

Type	Description
`ndarray`	2D error map, shape (H, W), values ≥ 0. Higher values = more likely anomalous.

Source code in app/pipelines/evaluation/scoring.py

def compute_pixel_error_map(
    original: np.ndarray, reconstruction: np.ndarray, alpha: float = 0.84, sigma: float = 2.0
) -> np.ndarray:
    """Compute the per-pixel absolute reconstruction error map.

    The error map is a weighted blend of Structural Similarity (SSIM) error
    and channel-wise Mean Absolute Error (MAE), aligned with the training loss.
    A Gaussian blur is applied to smooth noise and cluster anomaly predictions.

    Args:
        original: Original normalised image, shape (H, W, 3), values in [0, 1].
        reconstruction: Reconstructed image from the CAE, same shape as original.
        alpha: Weight for the SSIM component (default 0.84, matches loss).
        sigma: Standard deviation for Gaussian kernel (default 2.0).

    Returns:
        2D error map, shape (H, W), values ≥ 0. Higher values = more likely anomalous.
    """
    from scipy.ndimage import gaussian_filter
    from skimage.metrics import structural_similarity as ssim

    # 1. Structural Error Map (1 - SSIM)
    _, ssim_map = ssim(original, reconstruction, data_range=1.0, channel_axis=-1, full=True)  # type: ignore[no-untyped-call]
    ssim_error = 1.0 - np.mean(ssim_map, axis=-1)  # (H, W)

    # 2. Pixel Error Map (MAE)
    per_channel_error = np.abs(original - reconstruction)  # (H, W, 3)
    mae_error = np.mean(per_channel_error, axis=-1)  # (H, W)

    # 3. Blend them together
    error_map = alpha * ssim_error + (1.0 - alpha) * mae_error

    # 4. Smooth the result to cluster defect pixels
    if sigma > 0:
        error_map = gaussian_filter(error_map, sigma=sigma)

    return error_map

`top_k_pooling(error_map: np.ndarray, k: int | None = None, k_fraction: float = 0.002) -> float` ¶

Compute the image-level anomaly score using Top-K pooling.

Top-K pooling is significantly more robust than max-pooling because: - Single noisy pixels (sensor spikes, JPEG artefacts) produce 1 high pixel. - Real defects (scratches, cracks) produce a cluster of many high pixels. Averaging the top-K pixels dilutes isolated spikes while keeping defect clusters high.

Parameters:

Name	Type	Description	Default
`error_map`	`ndarray`	2D pixel error map, shape (H, W), values ≥ 0.	required
`k`	`int \| None`	Explicit number of top pixels to average. If None, derived from `k_fraction`.	`None`
`k_fraction`	`float`	Fraction of total pixels to use as K when `k` is not specified. Default 0.002 = 0.2% of pixels. For 128x128 → K ~= 33 pixels.	`0.002`

Returns:

Type	Description
`float`	Single float representing the image-level anomaly score. Higher = more anomalous.

Source code in app/pipelines/evaluation/scoring.py

def top_k_pooling(error_map: np.ndarray, k: int | None = None, k_fraction: float = 0.002) -> float:
    """Compute the image-level anomaly score using Top-K pooling.

    Top-K pooling is significantly more robust than max-pooling because:
    - Single noisy pixels (sensor spikes, JPEG artefacts) produce 1 high pixel.
    - Real defects (scratches, cracks) produce a *cluster* of many high pixels.
    Averaging the top-K pixels dilutes isolated spikes while keeping defect clusters high.

    Args:
        error_map: 2D pixel error map, shape (H, W), values ≥ 0.
        k: Explicit number of top pixels to average. If None, derived from ``k_fraction``.
        k_fraction: Fraction of total pixels to use as K when ``k`` is not specified.
            Default 0.002 = 0.2% of pixels. For 128x128 → K ~= 33 pixels.

    Returns:
        Single float representing the image-level anomaly score. Higher = more anomalous.
    """
    flat = error_map.flatten()
    total_pixels = len(flat)

    if k is None:
        k = max(1, int(total_pixels * k_fraction))

    # Sort descending and take the top-k
    top_k_values = np.partition(flat, -k)[-k:]  # np.partition is O(n), faster than full sort
    return float(np.mean(top_k_values))

`app.pipelines.evaluation.heatmaps` ¶

Reconstruction error heatmaps and explainability overlays for autoencoder models.

This module provides pixel-level reconstruction error calculation and visual overlay generation for autoencoder-based anomaly detection models (such as the fully convolutional Keras CAE and PyTorch Autoencoder).

Why Reconstruction Error Over Grad-CAM

Grad-CAM is an attribution technique originally designed for discriminative classification networks, where a scalar class score is backpropagated to intermediate convolutional feature maps to highlight the receptive fields influencing a decision. For autoencoder anomaly detection, direct pixel-wise reconstruction error is mathematically and practically superior:

Native Generative Objective: Autoencoders are trained exclusively on normal patterns to reconstruct nominal image geometry and texture. The anomaly signal is fundamentally defined as the residual between the original image and its reconstruction (e.g., squared pixel error or structural dissimilarity). Because the decoder yields a full-resolution spatial reconstruction directly, no gradient attribution proxy is needed.
Resolution and Spatial Locality: Backpropagating an aggregated scalar reconstruction loss via Grad-CAM pools gradients into coarse bottleneck feature maps (e.g., H/16 x W/16), yielding blurry, low-resolution saliency maps that require bilinear upsampling and can suffer from gradient saturation. In contrast, the per-pixel residual map preserves fine-grained defect contours at full spatial resolution without gradient artifacts.

To produce smooth, visually interpretable overlays analogous to Grad-CAM, the raw per-pixel reconstruction residual is regularized with a gentle Gaussian blur (default sigma=3.0), percentile-clipped to suppress outliers, and blended over the original image using a perceptual colormap.

Typical usage example

error_dict = compute_error_heatmap(model, input_image, sigma=3.0) overlay = overlay_heatmap(original_uint8_image, error_dict["heatmap"]) gt_overlay = overlay_ground_truth(overlay, binary_ground_truth_mask)

`compute_error_heatmap(model: Any, image: np.ndarray, sigma: float = 3.0, reconstruction: np.ndarray | None = None) -> dict[str, np.ndarray]` ¶

Compute a smoothed reconstruction error heatmap for a single image.

Parameters:

Name	Type	Description	Default
`model`	`Any`	A compiled `tf.keras.Model` produced by `build_cae()`.	required
`image`	`ndarray`	Single normalised image, shape (H, W, 3), float32 values in [0, 1].	required
`sigma`	`float`	Standard deviation for the Gaussian blur (smoothness).	`3.0`
`reconstruction`	`ndarray \| None`	Optional precomputed reconstructed image, shape (H, W, 3).	`None`

Returns:

Type	Description
`dict[str, ndarray]`	Dictionary containing: - `"heatmap"`: Normalised error heatmap, shape (H, W), float32 in [0, 1]. Higher values = regions with greater reconstruction error.

Source code in app/pipelines/evaluation/heatmaps.py

def compute_error_heatmap(
    model: Any,
    image: np.ndarray,
    sigma: float = 3.0,
    reconstruction: np.ndarray | None = None,
) -> dict[str, np.ndarray]:
    """Compute a smoothed reconstruction error heatmap for a single image.

    Args:
        model: A compiled ``tf.keras.Model`` produced by ``build_cae()``.
        image: Single normalised image, shape (H, W, 3), float32 values in [0, 1].
        sigma: Standard deviation for the Gaussian blur (smoothness).
        reconstruction: Optional precomputed reconstructed image, shape (H, W, 3).

    Returns:
        Dictionary containing:
            - ``"heatmap"``: Normalised error heatmap, shape (H, W), float32 in [0, 1].
                Higher values = regions with greater reconstruction error.
    """
    from app.pipelines.evaluation.scoring import compute_pixel_error_map

    # 1. Obtain reconstruction
    if reconstruction is None:
        image_batch = np.expand_dims(image, 0)  # (1, H, W, 3)
        reconstruction = model.predict(image_batch, verbose=0)[0]

    # 2. Pixel-wise error (matching the exact scoring logic)
    pixel_error = compute_pixel_error_map(image, reconstruction)

    # 3. Smooth with Gaussian filter for visual appeal
    heatmap = scipy.ndimage.gaussian_filter(pixel_error, sigma=sigma)

    # 4. Normalise robustly (1st-99th percentile) to [0, 1]
    p_low = float(np.percentile(heatmap, 1))
    p_high = float(np.percentile(heatmap, 99))

    if abs(p_high - p_low) > 1e-8:
        heatmap_norm = np.clip((heatmap - p_low) / (p_high - p_low), 0.0, 1.0)
    else:
        heatmap_norm = np.zeros_like(heatmap)

    logger.info(
        "Heatmap complete. Range: [%.4f, %.4f], Quantiles: [%.4f, %.4f]",
        heatmap.min(),
        heatmap.max(),
        p_low,
        p_high,
    )
    return {"heatmap": heatmap_norm.astype(np.float32)}

`overlay_ground_truth(original_image: np.ndarray, gt_mask: np.ndarray | None, color: tuple[int, int, int] = (0, 255, 0), alpha: float = 0.4) -> np.ndarray[Any, Any]` ¶

Blend a binary ground truth mask onto the original image.

Parameters:

Name	Type	Description	Default
`original_image`	`ndarray`	RGB image, shape (H, W, 3), uint8.	required
`gt_mask`	`ndarray \| None`	Binary mask, shape (H, W), uint8 (0 or 1). Can be None.	required
`color`	`tuple[int, int, int]`	RGB color tuple to draw the mask (e.g. Red=(255,0,0), Green=(0,255,0)).	`(0, 255, 0)`
`alpha`	`float`	Opacity of the mask overlay.	`0.4`

Returns:

Type	Description
`ndarray[Any, Any]`	RGB overlay image, shape (H, W, 3), uint8.

Source code in app/pipelines/evaluation/heatmaps.py

def overlay_ground_truth(
    original_image: np.ndarray,
    gt_mask: np.ndarray | None,
    color: tuple[int, int, int] = (0, 255, 0),  # Default to Green
    alpha: float = 0.4,
) -> np.ndarray[Any, Any]:
    """Blend a binary ground truth mask onto the original image.

    Args:
        original_image: RGB image, shape (H, W, 3), uint8.
        gt_mask: Binary mask, shape (H, W), uint8 (0 or 1). Can be None.
        color: RGB color tuple to draw the mask (e.g. Red=(255,0,0), Green=(0,255,0)).
        alpha: Opacity of the mask overlay.

    Returns:
        RGB overlay image, shape (H, W, 3), uint8.
    """
    if gt_mask is None:
        return original_image.copy()

    orig_norm = original_image.astype(np.float32) / 255.0
    color_norm = np.array(color, dtype=np.float32).reshape(1, 1, 3) / 255.0

    mask_3d = gt_mask[..., np.newaxis].astype(np.float32)  # (H, W, 1)

    # Where mask is 1, blend with color. Where 0, keep original.
    pixel_alpha = mask_3d * alpha

    blended = pixel_alpha * color_norm + (1.0 - pixel_alpha) * orig_norm
    blended = np.clip(blended, 0.0, 1.0)

    return (blended * 255).astype(np.uint8)  # type: ignore[no-any-return]

`overlay_heatmap(original_image: np.ndarray, heatmap: np.ndarray, alpha: float = 0.35, colormap: str = 'jet') -> np.ndarray[Any, Any]` ¶

Blend a heatmap onto the original image using a perceptual colourmap.

The heatmap is converted from greyscale → RGB via a colourmap (jet by default), then composited over the original image. Opacity is scaled per-pixel by the heatmap magnitude so regions with near-zero activation show the original image unchanged, while highly activated regions show a vivid colour tint.

Parameters:

Name	Type	Description	Default
`original_image`	`ndarray`	RGB image, shape (H, W, 3), uint8 values in [0, 255].	required
`heatmap`	`ndarray`	Normalised heatmap, shape (H, W), float32 in [0, 1].	required
`alpha`	`float`	Maximum overlay opacity for the highest-activation pixels. Default 0.35 keeps the original image clearly visible beneath the anomaly.	`0.35`
`colormap`	`str`	Matplotlib colourmap name applied to the heatmap.	`'jet'`

Returns:

Type	Description
`ndarray[Any, Any]`	RGB overlay image, shape (H, W, 3), uint8.

Source code in app/pipelines/evaluation/heatmaps.py

def overlay_heatmap(
    original_image: np.ndarray,
    heatmap: np.ndarray,
    alpha: float = 0.35,  # Reduced from 0.55 for more transparency (better visibility of the original part)
    colormap: str = "jet",
) -> np.ndarray[Any, Any]:
    """Blend a heatmap onto the original image using a perceptual colourmap.

    The heatmap is converted from greyscale → RGB via a colourmap (jet by default),
    then composited over the original image. Opacity is scaled per-pixel by the
    heatmap magnitude so regions with near-zero activation show the original image
    unchanged, while highly activated regions show a vivid colour tint.

    Args:
        original_image: RGB image, shape (H, W, 3), uint8 values in [0, 255].
        heatmap: Normalised heatmap, shape (H, W), float32 in [0, 1].
        alpha: Maximum overlay opacity for the highest-activation pixels.
            Default 0.35 keeps the original image clearly visible beneath the anomaly.
        colormap: Matplotlib colourmap name applied to the heatmap.

    Returns:
        RGB overlay image, shape (H, W, 3), uint8.
    """
    import matplotlib  # Lazy import

    cmap = matplotlib.colormaps[colormap]
    heatmap_rgb = cmap(heatmap)[..., :3]  # (H, W, 3), float64 in [0, 1]
    heatmap_rgb = heatmap_rgb.astype(np.float32)

    # Per-pixel alpha: proportional to heatmap magnitude
    pixel_alpha = (heatmap * alpha)[..., np.newaxis]  # (H, W, 1)

    orig_norm = original_image.astype(np.float32) / 255.0
    blended = pixel_alpha * heatmap_rgb + (1.0 - pixel_alpha) * orig_norm
    blended = np.clip(blended, 0.0, 1.0)

    return (blended * 255).astype(np.uint8)  # type: ignore[no-any-return]

User Interface & Reusable Components¶

`app.ui.components.selectors` ¶

Standardized Streamlit selectors for dataset directories and categories.

`render_dataset_and_category_selector(key_prefix: str, default_root: str = 'data/raw/mvtec_ad', default_category: str = 'bottle', on_category_change: Callable[..., Any] | None = None) -> tuple[str, str]` ¶

Render standardized two-column dataset root directory input and dynamic category selector.

Automatically scans the dataset root directory on disk to dynamically populate available category folders, while falling back cleanly to canonical MVTec AD benchmark categories.

Parameters:

Name	Type	Description	Default
`key_prefix`	`str`	Prefix used for Streamlit widget keys and session state (e.g. 'kcae', 'b', 'dino').	required
`default_root`	`str`	Default path string for the dataset root input.	`'data/raw/mvtec_ad'`
`default_category`	`str`	Default category name if available.	`'bottle'`
`on_category_change`	`Callable[..., Any] \| None`	Optional callback function triggered when the user changes the category.	`None`

Returns:

Type	Description
`tuple[str, str]`	A tuple of (dataset_root_directory_string, selected_category_string).

Source code in app/ui/components/selectors.py

def render_dataset_and_category_selector(
    key_prefix: str,
    default_root: str = "data/raw/mvtec_ad",
    default_category: str = "bottle",
    on_category_change: Callable[..., Any] | None = None,
) -> tuple[str, str]:
    """Render standardized two-column dataset root directory input and dynamic category selector.

    Automatically scans the dataset root directory on disk to dynamically populate
    available category folders, while falling back cleanly to canonical MVTec AD
    benchmark categories.

    Args:
        key_prefix: Prefix used for Streamlit widget keys and session state (e.g. 'kcae', 'b', 'dino').
        default_root: Default path string for the dataset root input.
        default_category: Default category name if available.
        on_category_change: Optional callback function triggered when the user changes the category.

    Returns:
        A tuple of (dataset_root_directory_string, selected_category_string).
    """
    root_key = f"{key_prefix}_root"
    cat_key = f"{key_prefix}_cat"

    st.session_state.setdefault(root_key, default_root)

    col1, col2 = st.columns(2)
    data_root = str(col1.text_input("Dataset Root Directory", key=root_key))

    categories = discover_dataset_categories(data_root)

    current_cat = st.session_state.get(cat_key, default_category)
    fallback_cat = default_category if default_category in categories else (categories[0] if categories else "bottle")
    if current_cat not in categories:
        st.session_state[cat_key] = fallback_cat
    else:
        st.session_state.setdefault(cat_key, fallback_cat)

    cat_index = categories.index(st.session_state[cat_key]) if st.session_state[cat_key] in categories else 0

    category = str(
        col2.selectbox(
            "Category Name",
            options=categories,
            index=cat_index,
            key=cat_key,
            on_change=on_category_change,
        )
    )

    return data_root, category

Core Configuration & Logging¶

`app.core.config` ¶

Validated runtime settings for the industrial component anomaly detection system.

`AppSettings` ¶

Bases: BaseSettings

Validated runtime settings for the application.

Attributes:

Name	Type	Description
`model_config`		Model configuration for pydantic-settings.
`PROJECT_NAME`	`str`	Name of the project.
`ENVIRONMENT`	`str`	Environment in which the application is running.
`API_V1_STR`	`str`	API version 1 string.
`DEBUG`	`bool`	Whether the application is running in debug mode.

Source code in app/core/config.py

class AppSettings(BaseSettings):
    """Validated runtime settings for the application.

    Attributes:
        model_config: Model configuration for pydantic-settings.
        PROJECT_NAME: Name of the project.
        ENVIRONMENT: Environment in which the application is running.
        API_V1_STR: API version 1 string.
        DEBUG: Whether the application is running in debug mode.
    """

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")

    PROJECT_NAME: str = Field(default="Python Project Template")
    ENVIRONMENT: str = Field(default="development")
    API_V1_STR: str = Field(default="/api/v1")
    DEBUG: bool = Field(default=False)

`app.core.logger` ¶

Logging utilities module.

`setup_logger(name: str = 'app') -> logging.Logger` ¶

Set up and configure standard logger emitting to stdout.

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the logger.	`'app'`

Returns:

Type	Description
`Logger`	Logger instance.

Source code in app/core/logger.py

def setup_logger(name: str = "app") -> logging.Logger:
    """Set up and configure standard logger emitting to stdout.

    Args:
        name: Name of the logger.

    Returns:
        Logger instance.
    """
    logger = logging.getLogger(name)
    if not logger.handlers:
        logger.setLevel(logging.INFO)
        handler = logging.StreamHandler(sys.stdout)
        formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
        handler.setFormatter(formatter)
        logger.addHandler(handler)
    return logger

API Reference¶

Application Domain & Data¶

app.domain.categories ¶

ANOMALY_DINO_MASKED_CATEGORIES: frozenset[str] = frozenset({'capsule', 'hazelnut', 'pill', 'screw', 'toothbrush'}) module-attribute ¶

MVTEC_CATEGORIES: tuple[str, ...] = tuple(sorted(MVTEC_OBJECT_CATEGORIES + MVTEC_TEXTURE_CATEGORIES)) module-attribute ¶

MVTEC_OBJECT_CATEGORIES: tuple[str, ...] = ('bottle', 'cable', 'capsule', 'hazelnut', 'metal_nut', 'pill', 'screw', 'toothbrush', 'transistor', 'zipper') module-attribute ¶

MVTEC_TEXTURE_CATEGORIES: tuple[str, ...] = ('carpet', 'grid', 'leather', 'tile', 'wood') module-attribute ¶

OBJECT_CATEGORIES: frozenset[str] = frozenset(MVTEC_OBJECT_CATEGORIES) module-attribute ¶

TEXTURE_CATEGORIES: frozenset[str] = frozenset(MVTEC_TEXTURE_CATEGORIES) module-attribute ¶

discover_dataset_categories(data_root: str | Path | None = None) -> list[str] ¶

app.domain.data ¶

FairEvaluationSplit dataclass ¶

fitting_paths: list[str] property ¶

test_paths: list[str] property ¶

validation_paths: list[str] property ¶

evidence() -> dict[str, Any] ¶

MVTecImageDataset ¶

__getitem__(index: int) -> tuple[Tensor, int, str] ¶

__init__(frame: pd.DataFrame, transform: Callable[[Image.Image], Tensor]) -> None ¶

__len__() -> int ¶

build_fair_evaluation_split(manifest: pd.DataFrame, category: str, *, validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION, seed: int = FAIR_EVALUATION_SPLIT_SEED) -> FairEvaluationSplit ¶

build_mvtec_manifest(root: str | Path) -> pd.DataFrame ¶

create_mvtec_dataset(manifest: pd.DataFrame, preprocessing_steps: list[dict[str, Any]] | None = None, image_size: tuple[int, int] = (256, 256)) -> MVTecImageDataset ¶

Command Line Interface¶

app.cli ¶

main() -> None ¶

preprocess_sys_argv() -> None ¶

Modelling Pipelines¶

app.pipelines.modelling.patchcore ¶

BaselineResult ¶

ConfusionMatrix dataclass ¶

EvaluationArtifacts dataclass ¶

__getitem__(index: int | slice) -> Any ¶

__iter__() -> Iterator[Any] ¶

__len__() -> int ¶

from_tuple(values: Any) -> EvaluationArtifacts classmethod ¶

FairEvaluationSplit dataclass ¶

fitting_paths: list[str] property ¶

test_paths: list[str] property ¶

validation_paths: list[str] property ¶

evidence() -> dict[str, Any] ¶

ImageEvaluationMetrics dataclass ¶

MetricLevelResult ¶

PixelEvaluationMetrics dataclass ¶

Thresholds dataclass ¶

build_fair_evaluation_split(manifest: pd.DataFrame, category: str, *, validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION, seed: int = FAIR_EVALUATION_SPLIT_SEED) -> FairEvaluationSplit ¶

build_mvtec_manifest(root: str | Path) -> pd.DataFrame ¶

delete_cached_patchcore_model(model_hash: str, registry_base: str | Path, soft_delete: bool = True) -> bool ¶

extract_and_save_pr_metrics(engine: Engine, model: Any, validation_dataloader: Any, test_dataloader: Any, base_dir: Path, run_heatmap: bool = False, model_name: str = 'PatchCore') -> EvaluationArtifacts ¶

list_trashed_patchcore_models(registry_base: str | Path) -> list[dict[str, Any]] ¶

purge_patchcore_trash(registry_base: str | Path, model_hash: str | None = None) -> int ¶

restore_cached_patchcore_model(model_hash: str, registry_base: str | Path) -> bool ¶

app.pipelines.modelling.patchcore.optuna_study ¶

objective(trial: optuna.Trial, category_name: str, data_root: str = 'data/raw/mvtec_ad') -> float ¶

run_study(category_name: str, n_trials: int = 30, data_root: str = 'data/raw/mvtec_ad') -> dict[str, Any] ¶

app.pipelines.modelling.keras_cae.cae_pipeline ¶

app.pipelines.modelling.keras_cae.optuna_study ¶

objective(trial: optuna.Trial, category_name: str, data_root: str = 'data/raw/mvtec_ad') -> float ¶

run_study(category_name: str, n_trials: int = 15, data_root: str = 'data/raw/mvtec_ad') -> dict[str, Any] ¶

app.pipelines.modelling.dino.v2 ¶

app.pipelines.modelling.dino.v3 ¶

app.pipelines.modelling.dino.enhanced ¶

EnhancedAnomalyDINOModel ¶

__init__(num_neighbours: int = 5, encoder_name: str = 'vit_small_patch14_dinov2', masking: bool = False, feature_layers: Sequence[int] = (8, 10, 11), position_radius: int = 1, spatial_weight: float = 0.05, density_neighbours: int = 5) -> None ¶

extract_features(image_tensor: torch.Tensor) -> torch.Tensor ¶

fit() -> None ¶

forward(input_tensor: torch.Tensor) -> torch.Tensor | InferenceBatch ¶

patchcore_image_score(neighbour_scores: torch.Tensor) -> torch.Tensor staticmethod ¶

app.pipelines.modelling.dino.engine ¶

compute_macro_average(category_results: dict[str, BaselineResult]) -> dict[str, float] ¶

instantiate_dino_model(num_neighbors: int, encoder_name: str, use_masking: bool, model_generation: Literal['dinov2', 'dinov3'], variant: DINOVariant, input_size: int, feature_layers: tuple[int, ...], position_radius: int, spatial_weight: float, density_neighbors: int) -> AnomalyDINO ¶

load_completed_category_result(base_dir: Path, category: str, model_hash: str, run_heatmap: bool, load_heatmaps: bool = True) -> BaselineResult | None ¶

release_accelerator_memory() -> None ¶

resolve_masking(masking: MaskingMode, category: str) -> bool ¶

save_all_category_summary(registry_base: Path | str, encoder_name: str, masking: MaskingMode, num_neighbors: int, category_results: dict[str, BaselineResult], macro_average: dict[str, float], model_name: str = 'DINO') -> None ¶

Preprocessing Subpackage¶

app.pipelines.preprocessing.factory ¶

build_pipeline_from_configs(configs: list[dict[str, Any]] | None) -> PreprocessingPipeline ¶

normalize_preprocessing_steps(steps: list[dict[str, Any]] | None) -> list[dict[str, Any]] ¶

app.pipelines.preprocessing.augmentation ¶

`app.domain.categories` ¶

`ANOMALY_DINO_MASKED_CATEGORIES: frozenset[str] = frozenset({'capsule', 'hazelnut', 'pill', 'screw', 'toothbrush'})` `module-attribute` ¶

`MVTEC_CATEGORIES: tuple[str, ...] = tuple(sorted(MVTEC_OBJECT_CATEGORIES + MVTEC_TEXTURE_CATEGORIES))` `module-attribute` ¶

`MVTEC_OBJECT_CATEGORIES: tuple[str, ...] = ('bottle', 'cable', 'capsule', 'hazelnut', 'metal_nut', 'pill', 'screw', 'toothbrush', 'transistor', 'zipper')` `module-attribute` ¶

`MVTEC_TEXTURE_CATEGORIES: tuple[str, ...] = ('carpet', 'grid', 'leather', 'tile', 'wood')` `module-attribute` ¶

`OBJECT_CATEGORIES: frozenset[str] = frozenset(MVTEC_OBJECT_CATEGORIES)` `module-attribute` ¶

`TEXTURE_CATEGORIES: frozenset[str] = frozenset(MVTEC_TEXTURE_CATEGORIES)` `module-attribute` ¶

`discover_dataset_categories(data_root: str | Path | None = None) -> list[str]` ¶

`app.domain.data` ¶

`FairEvaluationSplit` `dataclass` ¶

`fitting_paths: list[str]` `property` ¶

`test_paths: list[str]` `property` ¶

`validation_paths: list[str]` `property` ¶

`evidence() -> dict[str, Any]` ¶

`MVTecImageDataset` ¶

`getitem(index: int) -> tuple[Tensor, int, str]` ¶

`init(frame: pd.DataFrame, transform: Callable[[Image.Image], Tensor]) -> None` ¶

`len() -> int` ¶

`build_fair_evaluation_split(manifest: pd.DataFrame, category: str, *, validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION, seed: int = FAIR_EVALUATION_SPLIT_SEED) -> FairEvaluationSplit` ¶

`build_mvtec_manifest(root: str | Path) -> pd.DataFrame` ¶

`create_mvtec_dataset(manifest: pd.DataFrame, preprocessing_steps: list[dict[str, Any]] | None = None, image_size: tuple[int, int] = (256, 256)) -> MVTecImageDataset` ¶

`app.cli` ¶

`main() -> None` ¶

`preprocess_sys_argv() -> None` ¶

`app.pipelines.modelling.patchcore` ¶

`BaselineResult` ¶

`ConfusionMatrix` `dataclass` ¶

`EvaluationArtifacts` `dataclass` ¶

`getitem(index: int | slice) -> Any` ¶

`iter() -> Iterator[Any]` ¶

`len() -> int` ¶

`from_tuple(values: Any) -> EvaluationArtifacts` `classmethod` ¶

`FairEvaluationSplit` `dataclass` ¶

`fitting_paths: list[str]` `property` ¶

`test_paths: list[str]` `property` ¶

`validation_paths: list[str]` `property` ¶

`evidence() -> dict[str, Any]` ¶

`ImageEvaluationMetrics` `dataclass` ¶

`MetricLevelResult` ¶

`PixelEvaluationMetrics` `dataclass` ¶

`Thresholds` `dataclass` ¶

`build_fair_evaluation_split(manifest: pd.DataFrame, category: str, *, validation_fraction: float = FAIR_EVALUATION_VALIDATION_FRACTION, seed: int = FAIR_EVALUATION_SPLIT_SEED) -> FairEvaluationSplit` ¶

`build_mvtec_manifest(root: str | Path) -> pd.DataFrame` ¶

`delete_cached_patchcore_model(model_hash: str, registry_base: str | Path, soft_delete: bool = True) -> bool` ¶

`extract_and_save_pr_metrics(engine: Engine, model: Any, validation_dataloader: Any, test_dataloader: Any, base_dir: Path, run_heatmap: bool = False, model_name: str = 'PatchCore') -> EvaluationArtifacts` ¶

`list_trashed_patchcore_models(registry_base: str | Path) -> list[dict[str, Any]]` ¶

`purge_patchcore_trash(registry_base: str | Path, model_hash: str | None = None) -> int` ¶

`restore_cached_patchcore_model(model_hash: str, registry_base: str | Path) -> bool` ¶

`app.pipelines.modelling.patchcore.optuna_study` ¶

`objective(trial: optuna.Trial, category_name: str, data_root: str = 'data/raw/mvtec_ad') -> float` ¶

`run_study(category_name: str, n_trials: int = 30, data_root: str = 'data/raw/mvtec_ad') -> dict[str, Any]` ¶

`app.pipelines.modelling.keras_cae.cae_pipeline` ¶

`app.pipelines.modelling.keras_cae.optuna_study` ¶

`objective(trial: optuna.Trial, category_name: str, data_root: str = 'data/raw/mvtec_ad') -> float` ¶

`run_study(category_name: str, n_trials: int = 15, data_root: str = 'data/raw/mvtec_ad') -> dict[str, Any]` ¶

`app.pipelines.modelling.dino.v2` ¶

`app.pipelines.modelling.dino.v3` ¶

`app.pipelines.modelling.dino.enhanced` ¶

`EnhancedAnomalyDINOModel` ¶

`init(num_neighbours: int = 5, encoder_name: str = 'vit_small_patch14_dinov2', masking: bool = False, feature_layers: Sequence[int] = (8, 10, 11), position_radius: int = 1, spatial_weight: float = 0.05, density_neighbours: int = 5) -> None` ¶

`extract_features(image_tensor: torch.Tensor) -> torch.Tensor` ¶

`fit() -> None` ¶

`forward(input_tensor: torch.Tensor) -> torch.Tensor | InferenceBatch` ¶

`patchcore_image_score(neighbour_scores: torch.Tensor) -> torch.Tensor` `staticmethod` ¶

`app.pipelines.modelling.dino.engine` ¶

`compute_macro_average(category_results: dict[str, BaselineResult]) -> dict[str, float]` ¶

`instantiate_dino_model(num_neighbors: int, encoder_name: str, use_masking: bool, model_generation: Literal['dinov2', 'dinov3'], variant: DINOVariant, input_size: int, feature_layers: tuple[int, ...], position_radius: int, spatial_weight: float, density_neighbors: int) -> AnomalyDINO` ¶

`load_completed_category_result(base_dir: Path, category: str, model_hash: str, run_heatmap: bool, load_heatmaps: bool = True) -> BaselineResult | None` ¶

`release_accelerator_memory() -> None` ¶

`resolve_masking(masking: MaskingMode, category: str) -> bool` ¶

`save_all_category_summary(registry_base: Path | str, encoder_name: str, masking: MaskingMode, num_neighbors: int, category_results: dict[str, BaselineResult], macro_average: dict[str, float], model_name: str = 'DINO') -> None` ¶

`app.pipelines.preprocessing.factory` ¶

`build_pipeline_from_configs(configs: list[dict[str, Any]] | None) -> PreprocessingPipeline` ¶

`normalize_preprocessing_steps(steps: list[dict[str, Any]] | None) -> list[dict[str, Any]]` ¶

`app.pipelines.preprocessing.augmentation` ¶

`OBJECT_CATEGORIES: frozenset[str] = frozenset(MVTEC_OBJECT_CATEGORIES)` `module-attribute` ¶

`TEXTURE_CATEGORIES: frozenset[str] = frozenset(MVTEC_TEXTURE_CATEGORIES)` `module-attribute` ¶

`ObjectAugmenter` ¶

`call(image: Image.Image) -> Image.Image` ¶

`init(brightness_range: tuple[float, float] = (0.9, 1.1), contrast_range: tuple[float, float] = (0.9, 1.1), saturation_range: tuple[float, float] = (0.9, 1.1), noise_std: float = 0.02) -> None` ¶