mirror of
https://github.com/hanzoai/anomalib.git
synced 2026-10-30 18:48:11 +00:00
zeekay
dda5b2a367
The previous commit was meant to remove .github/workflows and removed every tracked path instead — 892 files, the whole repo. This puts back the 874 that were never in scope; the eighteen workflow files stay gone, for the reasons that commit gives. Content is byte-identical to 9d2aec9 outside .github/workflows.
70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
"""Test tiled prediction storage class."""
|
|
|
|
# Copyright (C) 2023-2025 Intel Corporation
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
import copy
|
|
from collections.abc import Callable
|
|
|
|
import torch
|
|
from torch import Tensor
|
|
|
|
from anomalib.data import AnomalibDataModule
|
|
from anomalib.pipelines.tiled_ensemble.components.utils.prediction_data import EnsemblePredictions
|
|
|
|
|
|
class TestPredictionData:
|
|
"""Test EnsemblePredictions class, used for tiled prediction storage."""
|
|
|
|
@staticmethod
|
|
def store_all(data: EnsemblePredictions, datamodule: AnomalibDataModule) -> dict:
|
|
"""Store the tiled predictions in the EnsemblePredictions object."""
|
|
tile_dict = {}
|
|
for tile_index in [(0, 0), (0, 1), (1, 0), (1, 1)]:
|
|
datamodule.external_collate_fn.tile_index = tile_index
|
|
|
|
tile_prediction = []
|
|
for batch in iter(datamodule.train_dataloader()):
|
|
# set mock maps to just one channel of image
|
|
batch.anomaly_maps = batch.image.clone()[:, 0, :, :].unsqueeze(1)
|
|
# set mock pred mask to mask but add channel
|
|
batch.pred_masks = batch.gt_mask.clone().unsqueeze(1)
|
|
tile_prediction.append(batch)
|
|
# save original
|
|
tile_dict[tile_index] = copy.deepcopy(tile_prediction)
|
|
# store to prediction storage object
|
|
data.add_tile_prediction(tile_index, tile_prediction)
|
|
|
|
return tile_dict
|
|
|
|
@staticmethod
|
|
def verify_equal(name: str, tile_dict: dict, storage: EnsemblePredictions, eq_funct: Callable) -> bool:
|
|
"""Verify that all data at same tile index and same batch index matches."""
|
|
batch_num = len(tile_dict[0, 0])
|
|
|
|
for batch_i in range(batch_num):
|
|
# batch is dict where key: tile index and val is batched data of that tile
|
|
curr_batch = storage.get_batch_tiles(batch_i)
|
|
|
|
# go over all indices of current batch of stored data
|
|
for tile_index, stored_data_batch in curr_batch.items():
|
|
stored_data = getattr(stored_data_batch, name)
|
|
# get original data dict at current tile index and batch index
|
|
original_data = getattr(tile_dict[tile_index][batch_i], name)
|
|
if isinstance(original_data, Tensor):
|
|
if not eq_funct(original_data, stored_data):
|
|
return False
|
|
elif original_data != stored_data:
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_prediction_object(self, get_datamodule: AnomalibDataModule) -> None:
|
|
"""Test prediction storage class."""
|
|
datamodule = get_datamodule
|
|
storage = EnsemblePredictions()
|
|
original = self.store_all(storage, datamodule)
|
|
|
|
for name in original[0, 0][0].__dict__:
|
|
assert self.verify_equal(name, original, storage, torch.equal), f"{name} doesn't match"
|