sensospot_parser/tests/test_normalisation.py

from collections import namedtuple

import pandas

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


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["Calc.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.Saturation": [4, 2, 3, 4]})

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

    assert list(result["Calc.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.utils import split_data_frame
    from sensospot_data.normalisation import _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 RAW_DATA_NORMALIZATION_MAP
    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 RAW_DATA_NORMALIZATION_MAP.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["Calc.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["Calc.Normalized.Spot.Mean"]) == expected_values


def test_normalize_measurement(data_frame_with_params):
    from sensospot_data.normalisation import split_channels

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

    result = split_channels(data_frame_with_params, exposure_map)
    cy3_df, cy5_df = result["Cy3"], result["Cy5"]

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