You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
294 lines
8.7 KiB
294 lines
8.7 KiB
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 ( |
|
_reduce_overflow_in_channel, |
|
_check_overflow_limit, |
|
) |
|
|
|
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 ( |
|
_reduce_overflow_in_channel, |
|
_check_overflow_limit, |
|
) |
|
|
|
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 ( |
|
_infer_normalization_map, |
|
_split_data_frame, |
|
) |
|
|
|
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_exposure(normalization_data_frame): |
|
from sensospot_data.normalisation import ( |
|
_normalize_exposure, |
|
reduce_overflow, |
|
) |
|
from sensospot_data.columns import COLUMN_NORMALIZATION |
|
|
|
reduced = reduce_overflow(normalization_data_frame, "Saturation", 1) |
|
result = _normalize_exposure(reduced["Cy5"], 100) |
|
|
|
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 ( |
|
normalize_exposure_time, |
|
reduce_overflow, |
|
) |
|
|
|
reduced = reduce_overflow(normalization_data_frame, "Saturation", 1) |
|
result = normalize_exposure_time(reduced, {"Cy5": 100, "Cy3": 0}) |
|
|
|
assert "Cy5" in result |
|
|
|
sorted_results = result["Cy5"].sort_values( |
|
by=["Well.Row", "Well.Column", "Pos.Id"] |
|
) |
|
expected_values = [2, 8, 30, 2, 20, 20, 20, 20, 200, 200, 200, 200] |
|
|
|
assert list(sorted_results["Normalized.Spot.Mean"]) == expected_values |
|
|
|
|
|
def test_normalize_exposure_time_infered_map(normalization_data_frame): |
|
from sensospot_data.normalisation import ( |
|
normalize_exposure_time, |
|
reduce_overflow, |
|
) |
|
|
|
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.normalisation import normalize_measurement |
|
from sensospot_data.parser import process_folder |
|
|
|
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), |
|
} |
|
normalization_map = {"Cy5": 25} |
|
|
|
result = normalize_measurement(data_frame, exposure_map, normalization_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() == 25
|
|
|