anomaly

AnomalibDetection(config, module_function, checkpoint_path=None, run_test=True, report=True)

Bases: Generic[AnomalyDataModuleT], LightningTask[AnomalyDataModuleT]

Anomaly Detection Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• module_function (DictConfig) –

  The function that instantiates the module and model

• checkpoint_path (Optional[str], default: None ) –

  The path to the checkpoint to load the model from. Defaults to None.

• run_test (bool, default: True ) –

  Whether to run the test after training. Defaults to False.

• report (bool, default: True ) –

  Whether to report the results. Defaults to False.

Source code in quadra/tasks/anomaly.py

def __init__(
    self,
    config: DictConfig,
    module_function: DictConfig,
    checkpoint_path: Optional[str] = None,
    run_test: bool = True,
    report: bool = True,
):
    super().__init__(
        config=config,
        checkpoint_path=checkpoint_path,
        run_test=run_test,
        report=report,
    )
    self._module: AnomalyModule
    self.module_function = module_function
    self.export_folder = "deployment_model"
    self.report_path = ""
    self.test_results: Optional[List[Dict]] = None

module: AnomalyModule property writable

Get the module.

export()

Export model for production.

Source code in quadra/tasks/anomaly.py

def export(self) -> None:
    """Export model for production."""
    if self.config.trainer.get("fast_dev_run"):
        log.warning("Skipping export since fast_dev_run is enabled")
        return

    model = self.module.model

    input_shapes = self.config.export.input_shapes

    half_precision = "16" in self.trainer.precision

    model_json, export_paths = export_model(
        config=self.config,
        model=model,
        export_folder=self.export_folder,
        half_precision=half_precision,
        input_shapes=input_shapes,
        idx_to_class={0: "good", 1: "defect"},
    )

    if len(export_paths) == 0:
        return

    model_json["image_threshold"] = np.round(self.module.image_threshold.value.item(), 3)
    model_json["pixel_threshold"] = np.round(self.module.pixel_threshold.value.item(), 3)
    model_json["anomaly_method"] = self.config.model.model.name

    with open(os.path.join(self.export_folder, "model.json"), "w") as f:
        json.dump(model_json, f, cls=utils.HydraEncoder)

generate_report()

Generate a report for the task and try to upload artifacts.

Source code in quadra/tasks/anomaly.py

def generate_report(self):
    """Generate a report for the task and try to upload artifacts."""
    self._generate_report()
    self._upload_artifacts()

prepare()

Prepare the task.

Source code in quadra/tasks/anomaly.py

def prepare(self) -> None:
    """Prepare the task."""
    super().prepare()
    self.module = self.config.model
    self.module.model = ModelSignatureWrapper(self.module.model)

test()

Lightning test.

Source code in quadra/tasks/anomaly.py

def test(self) -> Any:
    """Lightning test."""
    self.test_results = super().test()
    return self.test_results

AnomalibEvaluation(config, model_path, use_training_threshold=False, device=None, training_threshold_type=None)

Bases: Evaluation[AnomalyDataModule]

Evaluation task for Anomalib.

Parameters:

• config (DictConfig) –

  Task configuration

• model_path (str) –

  Path to the model folder that contains an exported model

• use_training_threshold (bool, default: False ) –

  Whether to use the training threshold for the evaluation or use the one that maximizes the F1 score on the test set.

• device (Optional[str], default: None ) –

  Device to use for evaluation. If None, the device is automatically determined.

Source code in quadra/tasks/anomaly.py

def __init__(
    self,
    config: DictConfig,
    model_path: str,
    use_training_threshold: bool = False,
    device: Optional[str] = None,
    training_threshold_type: Optional[Literal["image", "pixel"]] = None,
):
    super().__init__(config=config, model_path=model_path, device=device)

    self.use_training_threshold = use_training_threshold

    if training_threshold_type is not None and training_threshold_type not in ["image", "pixel"]:
        raise ValueError("Training threshold type must be either image or pixel")

    if training_threshold_type is None and use_training_threshold:
        log.warning("Using training threshold but no training threshold type is provided, defaulting to image")
        training_threshold_type = "image"

    self.training_threshold_type = training_threshold_type

execute()

Execute the evaluation.

Source code in quadra/tasks/anomaly.py

def execute(self) -> None:
    """Execute the evaluation."""
    self.prepare()
    self.test()
    self.generate_report()
    self.finalize()

generate_report()

Generate report.

Source code in quadra/tasks/anomaly.py

def generate_report(self) -> None:
    """Generate report."""
    log.info("Generating report")
    if len(self.report_path) > 0:
        os.makedirs(self.report_path, exist_ok=True)

    os.makedirs(os.path.join(self.report_path, "predictions"), exist_ok=True)
    os.makedirs(os.path.join(self.report_path, "heatmaps"), exist_ok=True)

    anomaly_scores = self.metadata["anomaly_scores"].cpu().numpy()
    good_scores = anomaly_scores[np.where(np.array(self.metadata["image_labels"]) == 0)]
    defect_scores = anomaly_scores[np.where(np.array(self.metadata["image_labels"]) == 1)]

    count_overlapping_scores = 0

    if len(good_scores) != 0 and len(defect_scores) != 0 and defect_scores.min() <= good_scores.max():
        count_overlapping_scores = len(
            np.where((anomaly_scores >= defect_scores.min()) & (anomaly_scores <= good_scores.max()))[0]
        )

    plot_cumulative_histogram(good_scores, defect_scores, self.metadata["threshold"], self.report_path)

    json_output = {
        "observations": [],
        "threshold": np.round(self.metadata["threshold"], 3),
        "f1_score": np.round(self.metadata["optimal_f1"], 3),
        "metrics": {
            "overlapping_scores": count_overlapping_scores,
        },
    }

    min_anomaly_score = self.metadata["anomaly_scores"].min().item()
    max_anomaly_score = self.metadata["anomaly_scores"].max().item()

    if min_anomaly_score == max_anomaly_score:
        # Handle the case where all anomaly scores are the same, skip normalization
        min_anomaly_score = 0
        max_anomaly_score = 1

    tg, fb, fg, tb = 0, 0, 0, 0

    mask_area = None
    crop_area = None

    if hasattr(self.datamodule, "valid_area_mask") and self.datamodule.valid_area_mask is not None:
        mask_area = cv2.imread(self.datamodule.valid_area_mask, 0)
        mask_area = (mask_area > 0).astype(np.uint8)

    if hasattr(self.datamodule, "crop_area") and self.datamodule.crop_area is not None:
        crop_area = self.datamodule.crop_area

    for img_path, gt_label, anomaly_score, anomaly_map in tqdm(
        zip(
            self.metadata["image_paths"],
            self.metadata["image_labels"],
            self.metadata["anomaly_scores"],
            self.metadata["anomaly_maps"],
        ),
        total=len(self.metadata["image_paths"]),
    ):
        img = cv2.imread(img_path, 0)
        if mask_area is not None:
            img = img * mask_area

        if crop_area is not None:
            img = img[crop_area[1] : crop_area[3], crop_area[0] : crop_area[2]]

        output_mask = (anomaly_map >= self.metadata["threshold"]).cpu().numpy().squeeze().astype(np.uint8)
        output_mask_name = os.path.splitext(os.path.basename(img_path))[0] + ".png"
        pred_label = int(anomaly_score >= self.metadata["threshold"])
        anomaly_probability = np.clip(
            ((anomaly_score.item() - self.metadata["threshold"]) / (max_anomaly_score - min_anomaly_score)) + 0.5,
            0,
            1,
        )

        json_output["observations"].append(
            {
                "image_path": os.path.dirname(img_path),
                "file_name": os.path.basename(img_path),
                "expectation": gt_label if gt_label != -1 else "",
                "prediction": pred_label,
                "prediction_mask": output_mask_name,
                "prediction_heatmap": output_mask_name,
                "is_correct": pred_label == gt_label if gt_label != -1 else True,
                "anomaly_score": f"{anomaly_score.item():.3f}",
                "anomaly_probability": f"{anomaly_probability:.3f}",
            }
        )

        if gt_label == 0 and pred_label == 0:
            tg += 1
        elif gt_label == 0 and pred_label == 1:
            fb += 1
        elif gt_label == 1 and pred_label == 0:
            fg += 1
        elif gt_label == 1 and pred_label == 1:
            tb += 1

        output_mask = output_mask * 255
        output_mask = cv2.resize(output_mask, (img.shape[1], img.shape[0]))
        cv2.imwrite(os.path.join(self.report_path, "predictions", output_mask_name), output_mask)

        # Normalize the heatmaps based on the current min and max anomaly score, otherwise even on good images the
        # anomaly map looks like there are defects while it's not true
        output_heatmap = (
            (anomaly_map.cpu().numpy().squeeze() - self.metadata["threshold"])
            / (max_anomaly_score - min_anomaly_score)
        ) + 0.5
        output_heatmap = np.clip(output_heatmap, 0, 1)
        output_heatmap = anomaly_map_to_color_map(output_heatmap, normalize=False)
        output_heatmap = cv2.resize(output_heatmap, (img.shape[1], img.shape[0]))
        cv2.imwrite(
            os.path.join(self.report_path, "heatmaps", output_mask_name),
            cv2.cvtColor(output_heatmap, cv2.COLOR_RGB2BGR),
        )

    json_output["metrics"]["confusion_matrix"] = {
        "class_labels": ["normal", "anomaly"],
        "matrix": [
            [tg, fb],
            [fg, tb],
        ],
    }

    with open(os.path.join(self.report_path, "anomaly_test_output.json"), "w") as f:
        json.dump(json_output, f)

prepare()

Prepare the evaluation.

Source code in quadra/tasks/anomaly.py

def prepare(self) -> None:
    """Prepare the evaluation."""
    super().prepare()
    self.datamodule = self.config.datamodule
    # prepare_data() must be explicitly called because there is no lightning training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="test")

test()

Perform test.

Source code in quadra/tasks/anomaly.py

@automatic_datamodule_batch_size(batch_size_attribute_name="test_batch_size")
def test(self) -> None:
    """Perform test."""
    log.info("Running test")
    test_dataloader = self.datamodule.test_dataloader()

    optimal_f1 = OptimalF1(num_classes=None, pos_label=1)  # type: ignore[arg-type]

    anomaly_scores = []
    anomaly_maps = []
    image_labels = []
    image_paths = []

    with torch.no_grad():
        for batch_item in tqdm(test_dataloader):
            batch_images = batch_item["image"]
            batch_labels = batch_item["label"]
            image_labels.extend(batch_labels.tolist())
            image_paths.extend(batch_item["image_path"])
            if self.model_data.get("anomaly_method") == "efficientad":
                model_output = self.deployment_model(batch_images.to(self.device), None)
            else:
                model_output = self.deployment_model(batch_images.to(self.device))
            anomaly_map, anomaly_score = model_output[0], model_output[1]
            anomaly_map = anomaly_map.cpu()
            anomaly_score = anomaly_score.cpu()
            known_labels = torch.where(batch_labels != -1)[0]
            if len(known_labels) > 0:
                # Skip computing F1 score for images without gt
                optimal_f1.update(anomaly_score[known_labels], batch_labels[known_labels])
            anomaly_scores.append(anomaly_score)
            anomaly_maps.append(anomaly_map)

    anomaly_scores = torch.cat(anomaly_scores)
    anomaly_maps = torch.cat(anomaly_maps)

    if any(x != -1 for x in image_labels):
        if self.use_training_threshold:
            _image_labels = torch.tensor(image_labels)
            threshold = torch.tensor(float(self.model_data[f"{self.training_threshold_type}_threshold"]))
            known_labels = torch.where(_image_labels != -1)[0]

            _image_labels = _image_labels[known_labels]
            _anomaly_scores = anomaly_scores[known_labels]

            pred_labels = (_anomaly_scores >= threshold).long()

            optimal_f1_score = torch.tensor(f1_score(_image_labels, pred_labels))
        else:
            optimal_f1_score = optimal_f1.compute()
            threshold = optimal_f1.threshold
    else:
        log.warning("No ground truth available during evaluation, use training image threshold for reporting")
        optimal_f1_score = torch.tensor(0)
        threshold = torch.tensor(float(self.model_data["image_threshold"]))

    log.info("Computed F1 score: %s", optimal_f1_score.item())
    self.metadata["anomaly_scores"] = anomaly_scores
    self.metadata["anomaly_maps"] = anomaly_maps
    self.metadata["image_labels"] = image_labels
    self.metadata["image_paths"] = image_paths
    self.metadata["threshold"] = threshold.item()
    self.metadata["optimal_f1"] = optimal_f1_score.item()