✨ feat(post-processing): Add MEBin post processor from AnomalyNCD

[rajeshgangireddy]

📝 Description

This is based on #2928
I have closed that PR and started this new one with additional changes.

✨ Changes

Select what type of change your PR is:

• 🚀 New feature (non-breaking change which adds functionality)
• 🐞 Bug fix (non-breaking change which fixes an issue)
• 🔄 Refactor (non-breaking change which refactors the code base)
• ⚡ Performance improvements
• 🎨 Style changes (code style/formatting)
• 🧪 Tests (adding/modifying tests)
• 📚 Documentation update
• 📦 Build system changes
• 🚧 CI/CD configuration
• 🔧 Chore (general maintenance)
• 🔒 Security update
• 💥 Breaking change (fix or feature that would cause existing functionality to not work as expected)

✅ Checklist

Before you submit your pull request, please make sure you have completed the following steps:

• 📚 I have made the necessary updates to the documentation (if applicable).
• 🧪 I have written tests that support my changes and prove that my fix is effective or my feature works (if applicable).
• 🏷️ My PR title follows conventional commit format.

For more information about code review checklists, see the Code Review Checklist.

[Copilot]

Pull Request Overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

Pull Request Overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

Pull Request Overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[rajeshgangireddy]

Thanks to @StarPlatinum7 for the contribution.

[rajeshgangireddy]

PS : I am yet to fully test this as a post processor with an with anomalib model.

[Copilot]

Pull Request Overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 4 comments.

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[StarPlatinum7]

Thanks for your review！This strategy is proposed in AnomalyNCD (CVPR2025).And I have test mebin as a post processor like this:

from anomalib.models import Patchcore
from anomalib.engine import Engine
from anomalib.post_processing import MEBinPostProcessor
datamodule = MVTecAD(
    category='bottle', # category in mvtec dataset
    root = "./mvtec_anomaly_detection", # path where dataset is stored
    train_batch_size = 16,
    eval_batch_size = 16,
)
engine = Engine(
    default_root_dir = './output', # path where results are saved

)
model = Patchcore(
    post_processor=MEBinPostProcessor(
        sample_rate=4,
        min_interval_len=4,
        erode=True
    )
)
engine.train(datamodule=datamodule, model=model)

[Copilot]

Pull Request Overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 3 comments.

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

Duplicate 'Example:' label on consecutive lines. Remove one of the duplicate labels.

Suggested change

	Example:
	>>> from anomalib.post_processing import MEBinPostProcessor
	>>> from anomalib.data import InferenceBatch
	>>> import torch
	>>> from anomalib.post_processing import MEBinPostProcessor
	>>> from anomalib.data import InferenceBatch
	>>> import torch
	>>> import torch

[Copilot]

[nitpick] The docstring contains a 'Stable Interval' definition that appears to be part of the description rather than a separate section. Consider integrating this into the main description or using a Note/Warning block for clarity.

Suggested change

	Stable Interval: A continuous threshold range in which the number of connected components remains constant,
	and the length of the threshold range is greater than or equal to the given length threshold
	(min_interval_len).
	A "stable interval" is defined as a continuous threshold range in which the number of connected components remains constant,
	and the length of the threshold range is greater than or equal to the given minimum interval length (`min_interval_len`).

[Copilot]

Type hint 'list' is not specific. Should be 'list[int]' to match the actual usage.

Suggested change

	anomaly_num_sequence (list): Sequence of connected component counts
	anomaly_num_sequence (list[int]): Sequence of connected component counts

[StarPlatinum7]

You can resolve the conflict by replacing the file tests/unit/post_processing/test_mebin_post_processor.py with the following code:

# SPDX-License-Identifier: Apache-2.0

"""Test the MEBinPostProcessor class."""

from unittest.mock import MagicMock, Mock, patch

import numpy as np
import pytest
import torch

from anomalib.data import InferenceBatch
from anomalib.post_processing import MEBinPostProcessor


class TestMEBinPostProcessor:
    """Test the MEBinPostProcessor class."""

    @staticmethod
    def test_initialization_default_params() -> None:
        """Test MEBinPostProcessor initialization with default parameters."""
        processor = MEBinPostProcessor()

        assert processor.sample_rate == 4
        assert processor.min_interval_len == 4
        assert processor.erode is True

    @staticmethod
    @pytest.mark.parametrize(
        ("sample_rate", "min_interval_len", "erode"),
        [
            (2, 3, True),
            (8, 6, False),
            (1, 1, True),
        ],
    )
    def test_initialization_custom_params(
        sample_rate: int,
        min_interval_len: int,
        erode: bool,
    ) -> None:
        """Test MEBinPostProcessor initialization with custom parameters."""
        processor = MEBinPostProcessor(
            sample_rate=sample_rate,
            min_interval_len=min_interval_len,
            erode=erode,
        )

        assert processor.sample_rate == sample_rate
        assert processor.min_interval_len == min_interval_len
        assert processor.erode == erode

    @staticmethod
    @patch("anomalib.post_processing.mebin_post_processor.MEBin")
    def test_forward_mebin_parameters(mock_mebin: MagicMock) -> None:
        """Test that MEBin is called with correct parameters."""
        # Setup mock
        mock_mebin_instance = Mock()
        mock_mebin_instance.binarize_anomaly_maps.return_value = (
            [np.array([[0, 0], [1, 1]], dtype=np.uint8)],
            [0.5],
        )
        mock_mebin.return_value = mock_mebin_instance

        # Create test data
        anomaly_maps = torch.rand(1, 1, 4, 4)
        predictions = InferenceBatch(
            pred_score=torch.tensor([0.8]),
            pred_label=torch.tensor([1]),
            anomaly_map=anomaly_maps,
            pred_mask=None,
        )

        # Test with custom parameters
        processor = MEBinPostProcessor(
            sample_rate=8,
            min_interval_len=6,
            erode=False,
        )
        _ = processor.forward(predictions)

        # Verify MEBin was called with correct parameters
        mock_mebin.assert_called_once_with(
            anomaly_map_list=mock_mebin.call_args[1]["anomaly_map_list"],
            sample_rate=8,
            min_interval_len=6,
            erode=False,
        )

    @staticmethod
    @patch("anomalib.post_processing.mebin_post_processor.MEBin")
    def test_forward_binary_mask_conversion(mock_mebin: MagicMock) -> None:
        """Test that binary masks are properly converted to 0/1 values."""
        # Setup mock to return masks with values > 0
        mock_mebin_instance = Mock()
        mock_mebin_instance.binarize_anomaly_maps.return_value = (
            [np.array([[0, 128], [255, 64]], dtype=np.uint8)],
            [0.5],
        )
        mock_mebin.return_value = mock_mebin_instance

        # Create test data
        anomaly_maps = torch.rand(1, 1, 2, 2)
        predictions = InferenceBatch(
            pred_score=torch.tensor([0.8]),
            pred_label=torch.tensor([1]),
            anomaly_map=anomaly_maps,
            pred_mask=None,
        )

        # Test forward pass
        processor = MEBinPostProcessor()
        result = processor.forward(predictions)

        # Verify that all values are either 0 or 1
        unique_values = torch.unique(result.pred_mask)
        assert torch.all((unique_values == 0) | (unique_values == 1))

    @staticmethod
    def test_forward_missing_anomaly_map() -> None:
        """Test that ValueError is raised when anomaly_map is None."""
        # Create test data without anomaly_map
        predictions = InferenceBatch(
            pred_score=torch.tensor([0.8]),
            pred_label=torch.tensor([1]),
            anomaly_map=None,
            pred_mask=None,
        )

        # Test forward pass should raise ValueError
        processor = MEBinPostProcessor()
        with pytest.raises(ValueError, match="Anomaly map is required for MEBin post-processing"):
            processor.forward(predictions)