sensospot_parser/tests/test_normailsation.py

from collections import namedtuple

import pandas
import pytest

from .conftest import EXAMPLE_DIR_WO_PARAMS, EXAMPLE_DIR_WITH_PARAMS

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


def test_split_data_frame(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import _split_data_frame

    data_frame = process_folder(example_dir / EXAMPLE_DIR_WITH_PARAMS)

    result = _split_data_frame(data_frame, "Well.Row")

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


def test_infer_exposure_from_parameters(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import _infer_exposure_from_parameters

    data_frame = process_folder(example_dir / EXAMPLE_DIR_WITH_PARAMS)
    result = _infer_exposure_from_parameters(data_frame)

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


def test_infer_exposure_from_parameters_raises_error(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import _infer_exposure_from_parameters

    data_frame = process_folder(example_dir / EXAMPLE_DIR_WO_PARAMS)

    with pytest.raises(ValueError) as excinfo:
        _infer_exposure_from_parameters(data_frame)

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


def test_apply_exposure_map(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import apply_exposure_map

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

    data_frame = process_folder(example_dir / EXAMPLE_DIR_WITH_PARAMS)
    result = apply_exposure_map(data_frame, 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(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import apply_exposure_map

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

    data_frame = process_folder(example_dir / EXAMPLE_DIR_WITH_PARAMS)

    with pytest.raises(ValueError) as excinfo:
        apply_exposure_map(data_frame, exposure_map)

    assert str(excinfo.value).startswith("Exposure Map differs")


def test_apply_exposure_map_from_parameters(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import apply_exposure_map

    data_frame = process_folder(example_dir / EXAMPLE_DIR_WITH_PARAMS)
    result = apply_exposure_map(data_frame, 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(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import apply_exposure_map

    data_frame = process_folder(example_dir / EXAMPLE_DIR_WO_PARAMS)

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

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


def test_check_overflow_limit_defaults():
    from sensospot_data.normalisation import _check_overflow_limit

    data_frame = pandas.DataFrame(data={"Spot.Mean": [0.1, 0.5, 0.6]})

    result = _check_overflow_limit(data_frame)

    assert list(result["Spot.Overflow"]) == [False, False, True]


def test_check_overflow_limit_custom_limit():
    from sensospot_data.normalisation import _check_overflow_limit

    data_frame = pandas.DataFrame(data={"Spot.Sat": [4, 2, 3, 4]})

    result = _check_overflow_limit(data_frame, "Spot.Sat", 2)

    assert list(result["Spot.Overflow"]) == [True, False, True, True]


def test_reduce_overflow_in_channel(normalization_data_frame):
    from sensospot_data.normalisation import (
        _check_overflow_limit,
        _reduce_overflow_in_channel,
    )

    data_frame = _check_overflow_limit(
        normalization_data_frame, "Saturation", 1
    )
    result = _reduce_overflow_in_channel(data_frame)

    sorted_results = result.sort_values(
        by=["Well.Row", "Well.Column", "Pos.Id"]
    )

    assert list(sorted_results["Value"]) == [
        1,
        2,
        3,
        1,
        10,
        10,
        10,
        10,
        100,
        100,
        100,
        100,
    ]


def test_reduce_overflow_in_channel_shortcut(normalization_data_frame):
    from sensospot_data.normalisation import (
        _check_overflow_limit,
        _reduce_overflow_in_channel,
    )

    normalization_data_frame["Exposure.Time"] = 1

    data_frame = _check_overflow_limit(
        normalization_data_frame, "Saturation", 1
    )
    result = _reduce_overflow_in_channel(data_frame)

    assert result is data_frame


def test_reduce_overflow(normalization_data_frame):
    from sensospot_data.normalisation import reduce_overflow

    result = reduce_overflow(normalization_data_frame, "Saturation", 1)

    assert "Cy5" in result

    sorted_results = result["Cy5"].sort_values(
        by=["Well.Row", "Well.Column", "Pos.Id"]
    )

    assert list(sorted_results["Value"]) == [
        1,
        2,
        3,
        1,
        10,
        10,
        10,
        10,
        100,
        100,
        100,
        100,
    ]


def test_infer_normalization_map(normalization_data_frame):
    from sensospot_data.normalisation import (
        _split_data_frame,
        _infer_normalization_map,
    )

    normalization_data_frame.loc[5, "Exposure.Channel"] = "Cy3"
    split_frames = _split_data_frame(
        normalization_data_frame, "Exposure.Channel"
    )

    result = _infer_normalization_map(split_frames)

    assert result == {"Cy3": 25, "Cy5": 50}


def test_normalize_channel(normalization_data_frame):
    from sensospot_data.columns import COLUMN_NORMALIZATION
    from sensospot_data.normalisation import (
        reduce_overflow,
        normalize_channel,
    )

    reduced = reduce_overflow(normalization_data_frame, "Saturation", 1)
    result = normalize_channel(reduced["Cy5"], 50)

    sorted_results = result.sort_values(
        by=["Well.Row", "Well.Column", "Pos.Id"]
    )
    expected_values = [2, 8, 30, 2, 20, 20, 20, 20, 200, 200, 200, 200]

    for normalized_col in COLUMN_NORMALIZATION.values():
        list(sorted_results[normalized_col]) == expected_values


def test_normalize_exposure_time(normalization_data_frame):
    from sensospot_data.normalisation import (
        reduce_overflow,
        normalize_exposure_time,
    )

    reduced = reduce_overflow(normalization_data_frame, "Saturation", 1)
    result = normalize_exposure_time(reduced)

    assert "Cy5" in result

    sorted_results = result["Cy5"].sort_values(
        by=["Well.Row", "Well.Column", "Pos.Id"]
    )
    expected_values = [1, 4, 15, 1, 10, 10, 10, 10, 100, 100, 100, 100]

    assert list(sorted_results["Normalized.Spot.Mean"]) == expected_values


def test_normalize_exposure_time_infered_map(normalization_data_frame):
    from sensospot_data.normalisation import (
        reduce_overflow,
        normalize_exposure_time,
    )

    reduced = reduce_overflow(normalization_data_frame, "Saturation", 1)
    result = normalize_exposure_time(reduced)

    assert "Cy5" in result

    sorted_results = result["Cy5"].sort_values(
        by=["Well.Row", "Well.Column", "Pos.Id"]
    )
    expected_values = [1, 4, 15, 1, 10, 10, 10, 10, 100, 100, 100, 100]

    assert list(sorted_results["Normalized.Spot.Mean"]) == expected_values


def test_normalize_measurement(example_dir):
    from sensospot_data.parser import process_folder
    from sensospot_data.normalisation import normalize_measurement

    sub_dir = example_dir / EXAMPLE_DIR_WITH_PARAMS
    data_frame = process_folder(sub_dir)

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

    result = normalize_measurement(data_frame, exposure_map)
    cy3_df, cy5_df = result["Cy3"], result["Cy5"]

    assert set(result.keys()) == {"Cy3", "Cy5"}
    assert cy3_df["Normalized.Exposure.Time"].unique() == 100
    assert cy5_df["Normalized.Exposure.Time"].unique() == 150