sensospot_parser/tests/test_utils.py

from collections import namedtuple

import pytest

ExposureSetting = namedtuple("ExposureSetting", ["channel", "time"])


def test_split_data_frame(data_frame_with_params):
    from sensospot_data.utils import split_data_frame

    result = split_data_frame(data_frame_with_params, "Well.Row")

    assert set(result.keys()) == set("ABC")
    for key, value_df in result.items():
        assert set(value_df["Well.Row"].unique()) == {key}


@pytest.mark.parametrize(
    "value,expected",
    [
        [[1, 2], True],
        [(1, 2), True],
        [{1, 2}, False],
        [{1: 2}, False],
        ["1, 2", False],
        [None, False],
    ],
)
def test_is_list_or_tuple(value, expected):
    from sensospot_data.utils import _is_list_or_tuple

    result = _is_list_or_tuple(value)

    assert result is expected


@pytest.mark.parametrize(
    "value,expected",
    [
        [1, True],
        [1.2, True],
        [{1, 2}, False],
        [{1: 2}, False],
        ["1", False],
        [None, False],
    ],
)
def test_is_numerical(value, expected):
    from sensospot_data.utils import _is_numerical

    result = _is_numerical(value)

    assert result is expected


def test_check_valid_exposure_map_entry_ok():
    from sensospot_data.utils import _check_valid_exposure_map_entry

    result = _check_valid_exposure_map_entry((2, 1))

    assert result is None


@pytest.mark.parametrize(
    "value", [[], [1], (1, 2, 3), {"a": 1, "b": 2}, ("A", "B")]
)
def test_check_valid_exposure_map_entry_raises_error(value):
    from sensospot_data.utils import _check_valid_exposure_map_entry

    with pytest.raises(ValueError):
        _check_valid_exposure_map_entry(value)


def test_check_exposure_map_ok(exposure_df):
    from sensospot_data.utils import _check_exposure_map

    exposure_map = {1: ("A", 10), 2: ("B", 20), 3: ("C", 30)}

    result = _check_exposure_map(exposure_df, exposure_map)

    assert result is None


def test_check_exposure_map_wrong_type(exposure_df):
    from sensospot_data.utils import _check_exposure_map

    exposure_map = []

    with pytest.raises(ValueError):
        _check_exposure_map(exposure_df, exposure_map)


def test_check_exposure_map_wrong_ids(exposure_df):
    from sensospot_data.utils import _check_exposure_map

    exposure_map = {1: ("A", 10), 2: ("B", 20), 4: ("D", 40)}

    with pytest.raises(ValueError):
        _check_exposure_map(exposure_df, exposure_map)


def test_check_exposure_map_invalid_entries(exposure_df):
    from sensospot_data.utils import _check_exposure_map

    exposure_map = {1: ("A", 10), 2: ("B", 20), 3: "ERROR"}

    with pytest.raises(ValueError):
        _check_exposure_map(exposure_df, exposure_map)


def test_infer_exposure_from_parameters(data_frame_with_params):
    from sensospot_data.utils import _set_exposure_data_from_parameters

    result = _set_exposure_data_from_parameters(data_frame_with_params)

    assert all(result["Exposure.Channel"] == result["Parameters.Channel"])
    assert all(result["Exposure.Time"] == result["Parameters.Time"])


def test_infer_exposure_from_parameters_raises_error(
    data_frame_without_params,
):
    from sensospot_data.utils import _set_exposure_data_from_parameters

    with pytest.raises(ValueError) as excinfo:
        _set_exposure_data_from_parameters(data_frame_without_params)

    assert str(excinfo.value).startswith("Exposure Map: measurement")


def test_apply_exposure_map(data_frame_with_params):
    from sensospot_data.utils import apply_exposure_map

    exposure_map = {
        1: ExposureSetting("Cy3", 100),
        2: ExposureSetting("Cy5", 15),
        3: ExposureSetting("Cy5", 150),
    }

    result = apply_exposure_map(data_frame_with_params, exposure_map)

    for key, value in exposure_map.items():
        mask = result["Exposure.Id"] == key
        partial = result.loc[mask]
        assert set(partial["Exposure.Channel"].unique()) == {value.channel}
        assert set(partial["Exposure.Time"].unique()) == {value.time}


def test_apply_exposure_map_raises_error(data_frame_with_params):
    from sensospot_data.utils import apply_exposure_map

    exposure_map = {
        1: ExposureSetting("Cy3", 100),
        2: ExposureSetting("Cy5", 15),
        "X": ExposureSetting("Cy5", 150),
    }

    with pytest.raises(ValueError):
        apply_exposure_map(data_frame_with_params, exposure_map)


def test_apply_exposure_map_from_parameters(data_frame_with_params):
    from sensospot_data.utils import apply_exposure_map

    result = apply_exposure_map(data_frame_with_params, None)

    assert all(result["Exposure.Channel"] == result["Parameters.Channel"])
    assert all(result["Exposure.Time"] == result["Parameters.Time"])


def test_apply_exposure_map_from_parameters_raises_error(
    data_frame_without_params,
):
    from sensospot_data.utils import apply_exposure_map

    with pytest.raises(ValueError) as excinfo:
        apply_exposure_map(data_frame_without_params, None)

    assert str(excinfo.value).startswith("Exposure Map: measurement")


def test_aggregate_defaults(normalization_data_frame):
    from sensospot_data.utils import aggregate

    normalization_data_frame.rename(
        columns={"Exposure.Time": "Exposure.Id"}, inplace=True
    )

    result = aggregate(normalization_data_frame, "Value", "median")

    assert result.columns == ["Aggregated.Median.Value"]
    assert result.index.names == ["Exposure.Id", "Well.Row", "Well.Column"]
    assert list(result["Aggregated.Median.Value"]) == [
        3,
        30,
        300,
        2,
        20,
        200,
        1,
        10,
        100,
    ]


def test_aggregate_on(normalization_data_frame):
    from sensospot_data.utils import aggregate

    result = aggregate(
        normalization_data_frame, "Value", "mean", on="Exposure.Time"
    )

    assert result.columns == ["Aggregated.Mean.Value"]
    assert result.index.names == ["Exposure.Time"]
    assert list(result["Aggregated.Mean.Value"]) == [111, 74, 37]


def test_aggregate_new_name(normalization_data_frame):
    from sensospot_data.utils import aggregate

    result = aggregate(
        normalization_data_frame,
        "Value",
        "mean",
        on="Exposure.Time",
        new_name="Foo",
    )

    assert result.columns == ["Foo"]
    assert result.index.names == ["Exposure.Time"]
    assert list(result["Foo"]) == [111, 74, 37]


def test_add_aggregate_new_name(normalization_data_frame):
    from sensospot_data.utils import add_aggregate

    result = add_aggregate(
        normalization_data_frame,
        "Value",
        "mean",
        on="Exposure.Time",
        new_name="Foo",
    )

    assert "Foo" in result.columns
    assert len(result.columns) == len(normalization_data_frame.columns) + 1
    assert result.index.names == [None]

    for exp, val in [(10, 111), (25, 74), (50, 37)]:
        mask = result["Exposure.Time"] == exp
        assert result.loc[mask, "Foo"].unique() == [val]