tests/test_attribution.py

"""Tests for modality attribution."""

from unittest.mock import MagicMock

import numpy as np
import pytest
import torch

from tests.conftest import make_segments


def _make_mock_model(n_vertices=100):
    """Create a mock model for attribution testing."""
    model = MagicMock()
    model.feature_dims = {"text": (2, 32), "audio": (2, 32), "video": (2, 32)}
    model.eval = MagicMock()

    call_count = [0]

    def fake_forward(batch, **kwargs):
        call_count[0] += 1
        B = 2
        # Make text-ablated predictions differ more to simulate importance
        if torch.all(batch.data.get("text", torch.ones(1)) == 0):
            return torch.ones(B, n_vertices, 10) * 0.5
        elif torch.all(batch.data.get("audio", torch.ones(1)) == 0):
            return torch.ones(B, n_vertices, 10) * 0.8
        elif torch.all(batch.data.get("video", torch.ones(1)) == 0):
            return torch.ones(B, n_vertices, 10) * 0.9
        return torch.ones(B, n_vertices, 10) * 1.0

    model.side_effect = fake_forward
    model.return_value = torch.ones(2, n_vertices, 10)
    # Override __call__ to use our function
    model.__class__ = type("MockModel", (), {
        "__call__": staticmethod(fake_forward),
        "feature_dims": {"text": (2, 32), "audio": (2, 32), "video": (2, 32)},
        "eval": lambda self: None,
    })
    # Simpler approach: just use a plain class
    class FakeModel:
        feature_dims = {"text": (2, 32), "audio": (2, 32), "video": (2, 32)}
        def eval(self): pass
        def __call__(self, batch, **kwargs):
            return fake_forward(batch, **kwargs)
    return FakeModel()


class TestModalityAttributor:
    def test_ablation_basic(self):
        from neuralset.dataloader import SegmentData

        from cortexlab.inference.attribution import ModalityAttributor

        model = _make_mock_model()
        attributor = ModalityAttributor(model)

        data = {
            "text": torch.randn(2, 2, 32, 20),
            "audio": torch.randn(2, 2, 32, 20),
            "video": torch.randn(2, 2, 32, 20),
            "subject_id": torch.zeros(2, dtype=torch.long),
        }
        batch = SegmentData(data=data, segments=make_segments(2))
        scores = attributor.attribute(batch)

        assert "text" in scores
        assert "audio" in scores
        assert "video" in scores
        assert scores["text"].shape == (100,)

    def test_text_most_important(self):
        from neuralset.dataloader import SegmentData

        from cortexlab.inference.attribution import ModalityAttributor

        model = _make_mock_model()
        attributor = ModalityAttributor(model)

        data = {
            "text": torch.randn(2, 2, 32, 20),
            "audio": torch.randn(2, 2, 32, 20),
            "video": torch.randn(2, 2, 32, 20),
            "subject_id": torch.zeros(2, dtype=torch.long),
        }
        batch = SegmentData(data=data, segments=make_segments(2))
        scores = attributor.attribute(batch)

        # Text ablation causes the biggest change (1.0 -> 0.5 = 0.5 diff)
        assert scores["text"].mean() > scores["audio"].mean()
        assert scores["audio"].mean() > scores["video"].mean()

    def test_normalised_scores_sum_to_one(self):
        from neuralset.dataloader import SegmentData

        from cortexlab.inference.attribution import ModalityAttributor

        model = _make_mock_model()
        attributor = ModalityAttributor(model)

        data = {
            "text": torch.randn(2, 2, 32, 20),
            "audio": torch.randn(2, 2, 32, 20),
            "video": torch.randn(2, 2, 32, 20),
            "subject_id": torch.zeros(2, dtype=torch.long),
        }
        batch = SegmentData(data=data, segments=make_segments(2))
        scores = attributor.attribute(batch)

        total = scores["text_normalised"] + scores["audio_normalised"] + scores["video_normalised"]
        np.testing.assert_allclose(total, 1.0, atol=1e-6)

    def test_with_roi_indices(self):
        from neuralset.dataloader import SegmentData

        from cortexlab.inference.attribution import ModalityAttributor

        roi_indices = {
            "V1": np.array([0, 1, 2, 3, 4]),
            "MT": np.array([10, 11, 12]),
        }
        model = _make_mock_model()
        attributor = ModalityAttributor(model, roi_indices=roi_indices)

        data = {
            "text": torch.randn(2, 2, 32, 20),
            "audio": torch.randn(2, 2, 32, 20),
            "video": torch.randn(2, 2, 32, 20),
            "subject_id": torch.zeros(2, dtype=torch.long),
        }
        batch = SegmentData(data=data, segments=make_segments(2))
        scores = attributor.attribute(batch)

        assert "text_roi" in scores
        assert "V1" in scores["text_roi"]
        assert "MT" in scores["text_roi"]