#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2018-2020 Satpy developers
#
# This file is part of satpy.
#
# satpy is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# satpy is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# satpy. If not, see <http://www.gnu.org/licenses/>.
"""Tests for compositors in composites/__init__.py."""
import os
import unittest
from datetime import datetime
from unittest import mock
import dask
import dask.array as da
import numpy as np
import pytest
import xarray as xr
from pyresample import AreaDefinition
import satpy
from satpy.tests.utils import CustomScheduler
# NOTE:
# The following fixtures are not defined in this file, but are used and injected by Pytest:
# - tmp_path
[docs]
class TestMatchDataArrays:
"""Test the utility method 'match_data_arrays'."""
[docs]
def _get_test_ds(self, shape=(50, 100), dims=("y", "x")):
"""Get a fake DataArray."""
from pyresample.geometry import AreaDefinition
data = da.random.random(shape, chunks=25)
area = AreaDefinition(
"test", "test", "test",
{"proj": "eqc", "lon_0": 0.0,
"lat_0": 0.0},
shape[dims.index("x")], shape[dims.index("y")],
(-20037508.34, -10018754.17, 20037508.34, 10018754.17))
attrs = {"area": area}
return xr.DataArray(data, dims=dims, attrs=attrs)
[docs]
def test_single_ds(self):
"""Test a single dataset is returned unharmed."""
from satpy.composites import CompositeBase
ds1 = self._get_test_ds()
comp = CompositeBase("test_comp")
ret_datasets = comp.match_data_arrays((ds1,))
assert ret_datasets[0].identical(ds1)
[docs]
def test_mult_ds_area(self):
"""Test multiple datasets successfully pass."""
from satpy.composites import CompositeBase
ds1 = self._get_test_ds()
ds2 = self._get_test_ds()
comp = CompositeBase("test_comp")
ret_datasets = comp.match_data_arrays((ds1, ds2))
assert ret_datasets[0].identical(ds1)
assert ret_datasets[1].identical(ds2)
[docs]
def test_mult_ds_no_area(self):
"""Test that all datasets must have an area attribute."""
from satpy.composites import CompositeBase
ds1 = self._get_test_ds()
ds2 = self._get_test_ds()
del ds2.attrs["area"]
comp = CompositeBase("test_comp")
with pytest.raises(ValueError, match="Missing 'area' attribute"):
comp.match_data_arrays((ds1, ds2))
[docs]
def test_mult_ds_diff_area(self):
"""Test that datasets with different areas fail."""
from pyresample.geometry import AreaDefinition
from satpy.composites import CompositeBase, IncompatibleAreas
ds1 = self._get_test_ds()
ds2 = self._get_test_ds()
ds2.attrs["area"] = AreaDefinition(
"test", "test", "test",
{"proj": "eqc", "lon_0": 0.0,
"lat_0": 0.0},
100, 50,
(-30037508.34, -20018754.17, 10037508.34, 18754.17))
comp = CompositeBase("test_comp")
with pytest.raises(IncompatibleAreas):
comp.match_data_arrays((ds1, ds2))
[docs]
def test_mult_ds_diff_dims(self):
"""Test that datasets with different dimensions still pass."""
from satpy.composites import CompositeBase
# x is still 50, y is still 100, even though they are in
# different order
ds1 = self._get_test_ds(shape=(50, 100), dims=("y", "x"))
ds2 = self._get_test_ds(shape=(3, 100, 50), dims=("bands", "x", "y"))
comp = CompositeBase("test_comp")
ret_datasets = comp.match_data_arrays((ds1, ds2))
assert ret_datasets[0].identical(ds1)
assert ret_datasets[1].identical(ds2)
[docs]
def test_mult_ds_diff_size(self):
"""Test that datasets with different sizes fail."""
from satpy.composites import CompositeBase, IncompatibleAreas
# x is 50 in this one, 100 in ds2
# y is 100 in this one, 50 in ds2
ds1 = self._get_test_ds(shape=(50, 100), dims=("x", "y"))
ds2 = self._get_test_ds(shape=(3, 50, 100), dims=("bands", "y", "x"))
comp = CompositeBase("test_comp")
with pytest.raises(IncompatibleAreas):
comp.match_data_arrays((ds1, ds2))
[docs]
def test_nondimensional_coords(self):
"""Test the removal of non-dimensional coordinates when compositing."""
from satpy.composites import CompositeBase
ds = self._get_test_ds(shape=(2, 2))
ds["acq_time"] = ("y", [0, 1])
comp = CompositeBase("test_comp")
ret_datasets = comp.match_data_arrays([ds, ds])
assert "acq_time" not in ret_datasets[0].coords
[docs]
def test_almost_equal_geo_coordinates(self):
"""Test that coordinates that are almost-equal still match.
See https://github.com/pytroll/satpy/issues/2668 for discussion.
Various operations like cropping and resampling can cause
geo-coordinates (y, x) to be very slightly unequal due to floating
point precision. This test makes sure that even in those cases we
can still generate composites from DataArrays with these coordinates.
"""
from satpy.composites import CompositeBase
from satpy.resample import add_crs_xy_coords
comp = CompositeBase("test_comp")
data_arr1 = self._get_test_ds(shape=(2, 2))
data_arr1 = add_crs_xy_coords(data_arr1, data_arr1.attrs["area"])
data_arr2 = self._get_test_ds(shape=(2, 2))
data_arr2 = data_arr2.assign_coords(
x=data_arr1.coords["x"] + 0.000001,
y=data_arr1.coords["y"],
crs=data_arr1.coords["crs"],
)
# data_arr2 = add_crs_xy_coords(data_arr2, data_arr2.attrs["area"])
# data_arr2.assign_coords(x=data_arr2.coords["x"].copy() + 1.1)
# default xarray alignment would fail and collapse one of our dims
assert 0 in (data_arr2 - data_arr1).shape
new_data_arr1, new_data_arr2 = comp.match_data_arrays([data_arr1, data_arr2])
assert 0 not in new_data_arr1.shape
assert 0 not in new_data_arr2.shape
assert 0 not in (new_data_arr2 - new_data_arr1).shape
[docs]
class TestRatioSharpenedCompositors:
"""Test RatioSharpenedRGB and SelfSharpendRGB compositors."""
[docs]
def setup_method(self):
"""Create test data."""
from pyresample.geometry import AreaDefinition
area = AreaDefinition("test", "test", "test",
{"proj": "merc"}, 2, 2,
(-2000, -2000, 2000, 2000))
attrs = {"area": area,
"start_time": datetime(2018, 1, 1, 18),
"modifiers": tuple(),
"resolution": 1000,
"calibration": "reflectance",
"units": "%",
"name": "test_vis"}
low_res_data = np.ones((2, 2), dtype=np.float64) + 4
low_res_data[1, 1] = 0.0 # produces infinite ratio
ds1 = xr.DataArray(da.from_array(low_res_data, chunks=2),
attrs=attrs, dims=("y", "x"),
coords={"y": [0, 1], "x": [0, 1]})
self.ds1 = ds1
ds2 = xr.DataArray(da.ones((2, 2), chunks=2, dtype=np.float64) + 2,
attrs=attrs, dims=("y", "x"),
coords={"y": [0, 1], "x": [0, 1]})
ds2.attrs["name"] += "2"
self.ds2 = ds2
ds3 = xr.DataArray(da.ones((2, 2), chunks=2, dtype=np.float64) + 3,
attrs=attrs, dims=("y", "x"),
coords={"y": [0, 1], "x": [0, 1]})
ds3.attrs["name"] += "3"
self.ds3 = ds3
# high resolution version
high_res_data = np.ones((2, 2), dtype=np.float64)
high_res_data[1, 0] = np.nan # invalid value in one band
ds4 = xr.DataArray(da.from_array(high_res_data, chunks=2),
attrs=attrs, dims=("y", "x"),
coords={"y": [0, 1], "x": [0, 1]})
ds4.attrs["name"] += "4"
ds4.attrs["resolution"] = 500
self.ds4 = ds4
# high resolution version - but too big
ds4_big = xr.DataArray(da.ones((4, 4), chunks=2, dtype=np.float64),
attrs=attrs.copy(), dims=("y", "x"),
coords={"y": [0, 1, 2, 3], "x": [0, 1, 2, 3]})
ds4_big.attrs["name"] += "4"
ds4_big.attrs["resolution"] = 500
ds4_big.attrs["rows_per_scan"] = 1
ds4_big.attrs["area"] = AreaDefinition("test", "test", "test",
{"proj": "merc"}, 4, 4,
(-2000, -2000, 2000, 2000))
self.ds4_big = ds4_big
[docs]
@pytest.mark.parametrize(
"init_kwargs",
[
{"high_resolution_band": "bad", "neutral_resolution_band": "red"},
{"high_resolution_band": "red", "neutral_resolution_band": "bad"}
]
)
def test_bad_colors(self, init_kwargs):
"""Test that only valid band colors can be provided."""
from satpy.composites import RatioSharpenedRGB
with pytest.raises(ValueError, match="RatioSharpenedRGB..*_band must be one of .*"):
RatioSharpenedRGB(name="true_color", **init_kwargs)
[docs]
def test_match_data_arrays(self):
"""Test that all areas have to be the same resolution."""
from satpy.composites import IncompatibleAreas, RatioSharpenedRGB
comp = RatioSharpenedRGB(name="true_color")
with pytest.raises(IncompatibleAreas):
comp((self.ds1, self.ds2, self.ds3), optional_datasets=(self.ds4_big,))
[docs]
def test_more_than_three_datasets(self):
"""Test that only 3 datasets can be passed."""
from satpy.composites import RatioSharpenedRGB
comp = RatioSharpenedRGB(name="true_color")
with pytest.raises(ValueError, match="Expected 3 datasets, got 4"):
comp((self.ds1, self.ds2, self.ds3, self.ds1), optional_datasets=(self.ds4_big,))
[docs]
def test_self_sharpened_no_high_res(self):
"""Test for exception when no high_res band is specified."""
from satpy.composites import SelfSharpenedRGB
comp = SelfSharpenedRGB(name="true_color", high_resolution_band=None)
with pytest.raises(ValueError, match="SelfSharpenedRGB requires at least one high resolution band, not 'None'"):
comp((self.ds1, self.ds2, self.ds3))
[docs]
def test_basic_no_high_res(self):
"""Test that three datasets can be passed without optional high res."""
from satpy.composites import RatioSharpenedRGB
comp = RatioSharpenedRGB(name="true_color")
res = comp((self.ds1, self.ds2, self.ds3))
assert res.shape == (3, 2, 2)
[docs]
def test_basic_no_sharpen(self):
"""Test that color None does no sharpening."""
from satpy.composites import RatioSharpenedRGB
comp = RatioSharpenedRGB(name="true_color", high_resolution_band=None)
res = comp((self.ds1, self.ds2, self.ds3), optional_datasets=(self.ds4,))
assert res.shape == (3, 2, 2)
[docs]
@pytest.mark.parametrize(
("high_resolution_band", "neutral_resolution_band", "exp_r", "exp_g", "exp_b"),
[
("red", None,
np.array([[1.0, 1.0], [np.nan, 1.0]], dtype=np.float64),
np.array([[0.6, 0.6], [np.nan, 3.0]], dtype=np.float64),
np.array([[0.8, 0.8], [np.nan, 4.0]], dtype=np.float64)),
("red", "green",
np.array([[1.0, 1.0], [np.nan, 1.0]], dtype=np.float64),
np.array([[3.0, 3.0], [np.nan, 3.0]], dtype=np.float64),
np.array([[0.8, 0.8], [np.nan, 4.0]], dtype=np.float64)),
("green", None,
np.array([[5 / 3, 5 / 3], [np.nan, 0.0]], dtype=np.float64),
np.array([[1.0, 1.0], [np.nan, 1.0]], dtype=np.float64),
np.array([[4 / 3, 4 / 3], [np.nan, 4 / 3]], dtype=np.float64)),
("green", "blue",
np.array([[5 / 3, 5 / 3], [np.nan, 0.0]], dtype=np.float64),
np.array([[1.0, 1.0], [np.nan, 1.0]], dtype=np.float64),
np.array([[4.0, 4.0], [np.nan, 4.0]], dtype=np.float64)),
("blue", None,
np.array([[1.25, 1.25], [np.nan, 0.0]], dtype=np.float64),
np.array([[0.75, 0.75], [np.nan, 0.75]], dtype=np.float64),
np.array([[1.0, 1.0], [np.nan, 1.0]], dtype=np.float64)),
("blue", "red",
np.array([[5.0, 5.0], [np.nan, 0.0]], dtype=np.float64),
np.array([[0.75, 0.75], [np.nan, 0.75]], dtype=np.float64),
np.array([[1.0, 1.0], [np.nan, 1.0]], dtype=np.float64))
]
)
def test_ratio_sharpening(self, high_resolution_band, neutral_resolution_band, exp_r, exp_g, exp_b):
"""Test RatioSharpenedRGB by different groups of high_resolution_band and neutral_resolution_band."""
from satpy.composites import RatioSharpenedRGB
comp = RatioSharpenedRGB(name="true_color", high_resolution_band=high_resolution_band,
neutral_resolution_band=neutral_resolution_band)
res = comp((self.ds1, self.ds2, self.ds3), optional_datasets=(self.ds4,))
assert "units" not in res.attrs
assert isinstance(res, xr.DataArray)
assert isinstance(res.data, da.Array)
data = res.values
np.testing.assert_allclose(data[0], exp_r, rtol=1e-5)
np.testing.assert_allclose(data[1], exp_g, rtol=1e-5)
np.testing.assert_allclose(data[2], exp_b, rtol=1e-5)
[docs]
@pytest.mark.parametrize(
("exp_shape", "exp_r", "exp_g", "exp_b"),
[
((3, 2, 2),
np.array([[5.0, 5.0], [5.0, 0]], dtype=np.float64),
np.array([[4.0, 4.0], [4.0, 0]], dtype=np.float64),
np.array([[16 / 3, 16 / 3], [16 / 3, 0]], dtype=np.float64))
]
)
def test_self_sharpened_basic(self, exp_shape, exp_r, exp_g, exp_b):
"""Test that three datasets can be passed without optional high res."""
from satpy.composites import SelfSharpenedRGB
comp = SelfSharpenedRGB(name="true_color")
res = comp((self.ds1, self.ds2, self.ds3))
data = res.values
assert data.shape == exp_shape
np.testing.assert_allclose(data[0], exp_r, rtol=1e-5)
np.testing.assert_allclose(data[1], exp_g, rtol=1e-5)
np.testing.assert_allclose(data[2], exp_b, rtol=1e-5)
[docs]
class TestDifferenceCompositor(unittest.TestCase):
"""Test case for the difference compositor."""
[docs]
def setUp(self):
"""Create test data."""
from pyresample.geometry import AreaDefinition
area = AreaDefinition("test", "test", "test",
{"proj": "merc"}, 2, 2,
(-2000, -2000, 2000, 2000))
attrs = {"area": area,
"start_time": datetime(2018, 1, 1, 18),
"modifiers": tuple(),
"resolution": 1000,
"name": "test_vis"}
ds1 = xr.DataArray(da.ones((2, 2), chunks=2, dtype=np.float64),
attrs=attrs, dims=("y", "x"),
coords={"y": [0, 1], "x": [0, 1]})
self.ds1 = ds1
ds2 = xr.DataArray(da.ones((2, 2), chunks=2, dtype=np.float64) + 2,
attrs=attrs, dims=("y", "x"),
coords={"y": [0, 1], "x": [0, 1]})
ds2.attrs["name"] += "2"
self.ds2 = ds2
# high res version
ds2 = xr.DataArray(da.ones((4, 4), chunks=2, dtype=np.float64) + 4,
attrs=attrs.copy(), dims=("y", "x"),
coords={"y": [0, 1, 2, 3], "x": [0, 1, 2, 3]})
ds2.attrs["name"] += "2"
ds2.attrs["resolution"] = 500
ds2.attrs["rows_per_scan"] = 1
ds2.attrs["area"] = AreaDefinition("test", "test", "test",
{"proj": "merc"}, 4, 4,
(-2000, -2000, 2000, 2000))
self.ds2_big = ds2
[docs]
def test_basic_diff(self):
"""Test that a basic difference composite works."""
from satpy.composites import DifferenceCompositor
comp = DifferenceCompositor(name="diff", standard_name="temperature_difference")
res = comp((self.ds1, self.ds2))
np.testing.assert_allclose(res.values, -2)
assert res.attrs.get("standard_name") == "temperature_difference"
[docs]
def test_bad_areas_diff(self):
"""Test that a difference where resolutions are different fails."""
from satpy.composites import DifferenceCompositor, IncompatibleAreas
comp = DifferenceCompositor(name="diff")
# too many arguments
with pytest.raises(ValueError, match="Expected 2 datasets, got 3"):
comp((self.ds1, self.ds2, self.ds2_big))
# different resolution
with pytest.raises(IncompatibleAreas):
comp((self.ds1, self.ds2_big))
[docs]
@pytest.fixture()
def fake_area():
"""Return a fake 2×2 area."""
from pyresample.geometry import create_area_def
return create_area_def("skierffe", 4087, area_extent=[-5_000, -5_000, 5_000, 5_000], shape=(2, 2))
[docs]
@pytest.fixture()
def fake_dataset_pair(fake_area):
"""Return a fake pair of 2×2 datasets."""
ds1 = xr.DataArray(da.full((2, 2), 8, chunks=2, dtype=np.float32), attrs={"area": fake_area})
ds2 = xr.DataArray(da.full((2, 2), 4, chunks=2, dtype=np.float32), attrs={"area": fake_area})
return (ds1, ds2)
[docs]
def test_ratio_compositor(fake_dataset_pair):
"""Test the ratio compositor."""
from satpy.composites import RatioCompositor
comp = RatioCompositor(name="ratio", standard_name="channel_ratio")
res = comp(fake_dataset_pair)
np.testing.assert_allclose(res.values, 2)
[docs]
def test_sum_compositor(fake_dataset_pair):
"""Test the sum compositor."""
from satpy.composites import SumCompositor
comp = SumCompositor(name="sum", standard_name="channel_sum")
res = comp(fake_dataset_pair)
np.testing.assert_allclose(res.values, 12)
[docs]
class TestDayNightCompositor(unittest.TestCase):
"""Test DayNightCompositor."""
[docs]
def setUp(self):
"""Create test data."""
bands = ["R", "G", "B"]
start_time = datetime(2018, 1, 1, 18, 0, 0)
# RGB
a = np.zeros((3, 2, 2), dtype=np.float32)
a[:, 0, 0] = 0.1
a[:, 0, 1] = 0.2
a[:, 1, 0] = 0.3
a[:, 1, 1] = 0.4
a = da.from_array(a, a.shape)
self.data_a = xr.DataArray(a, attrs={"test": "a", "start_time": start_time},
coords={"bands": bands}, dims=("bands", "y", "x"))
b = np.zeros((3, 2, 2), dtype=np.float32)
b[:, 0, 0] = np.nan
b[:, 0, 1] = 0.25
b[:, 1, 0] = 0.50
b[:, 1, 1] = 0.75
b = da.from_array(b, b.shape)
self.data_b = xr.DataArray(b, attrs={"test": "b", "start_time": start_time},
coords={"bands": bands}, dims=("bands", "y", "x"))
sza = np.array([[80., 86.], [94., 100.]], dtype=np.float32)
sza = da.from_array(sza, sza.shape)
self.sza = xr.DataArray(sza, dims=("y", "x"))
# fake area
my_area = AreaDefinition(
"test", "", "",
"+proj=longlat",
2, 2,
(-95.0, 40.0, -92.0, 43.0),
)
self.data_a.attrs["area"] = my_area
self.data_b.attrs["area"] = my_area
# not used except to check that it matches the data arrays
self.sza.attrs["area"] = my_area
[docs]
def test_daynight_sza(self):
"""Test compositor with both day and night portions when SZA data is included."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="day_night")
res = comp((self.data_a, self.data_b, self.sza))
res = res.compute()
expected = np.array([[0., 0.22122374], [0.5, 1.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected, rtol=1e-6)
[docs]
def test_daynight_area(self):
"""Test compositor both day and night portions when SZA data is not provided."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="day_night")
res = comp((self.data_a, self.data_b))
res = res.compute()
expected_channel = np.array([[0., 0.33164983], [0.66835017, 1.]], dtype=np.float32)
assert res.dtype == np.float32
for i in range(3):
np.testing.assert_allclose(res.values[i], expected_channel)
[docs]
def test_night_only_sza_with_alpha(self):
"""Test compositor with night portion with alpha band when SZA data is included."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="night_only", include_alpha=True)
res = comp((self.data_b, self.sza))
res = res.compute()
expected_red_channel = np.array([[np.nan, 0.], [0.5, 1.]], dtype=np.float32)
expected_alpha = np.array([[0., 0.3329599], [1., 1.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected_red_channel)
np.testing.assert_allclose(res.values[-1], expected_alpha)
[docs]
def test_night_only_sza_without_alpha(self):
"""Test compositor with night portion without alpha band when SZA data is included."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="night_only", include_alpha=False)
res = comp((self.data_a, self.sza))
res = res.compute()
expected = np.array([[0., 0.11042609], [0.6683502, 1.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected)
assert "A" not in res.bands
[docs]
def test_night_only_area_with_alpha(self):
"""Test compositor with night portion with alpha band when SZA data is not provided."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="night_only", include_alpha=True)
res = comp((self.data_b,))
res = res.compute()
expected_l_channel = np.array([[np.nan, 0.], [0.5, 1.]], dtype=np.float32)
expected_alpha = np.array([[np.nan, 0.], [0., 0.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected_l_channel)
np.testing.assert_allclose(res.values[-1], expected_alpha)
[docs]
def test_night_only_area_without_alpha(self):
"""Test compositor with night portion without alpha band when SZA data is not provided."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="night_only", include_alpha=False)
res = comp((self.data_b,))
res = res.compute()
expected = np.array([[np.nan, 0.], [0., 0.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected)
assert "A" not in res.bands
[docs]
def test_day_only_sza_with_alpha(self):
"""Test compositor with day portion with alpha band when SZA data is included."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="day_only", include_alpha=True)
res = comp((self.data_a, self.sza))
res = res.compute()
expected_red_channel = np.array([[0., 0.33164983], [0.66835017, 1.]], dtype=np.float32)
expected_alpha = np.array([[1., 0.6670401], [0., 0.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected_red_channel)
np.testing.assert_allclose(res.values[-1], expected_alpha)
[docs]
def test_day_only_sza_without_alpha(self):
"""Test compositor with day portion without alpha band when SZA data is included."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="day_only", include_alpha=False)
res = comp((self.data_a, self.sza))
res = res.compute()
expected_channel_data = np.array([[0., 0.22122373], [0., 0.]], dtype=np.float32)
assert res.dtype == np.float32
for i in range(3):
np.testing.assert_allclose(res.values[i], expected_channel_data)
assert "A" not in res.bands
[docs]
def test_day_only_area_with_alpha(self):
"""Test compositor with day portion with alpha_band when SZA data is not provided."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="day_only", include_alpha=True)
res = comp((self.data_a,))
res = res.compute()
expected_l_channel = np.array([[0., 0.33164983], [0.66835017, 1.]], dtype=np.float32)
expected_alpha = np.array([[1., 1.], [1., 1.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected_l_channel)
np.testing.assert_allclose(res.values[-1], expected_alpha)
[docs]
def test_day_only_area_with_alpha_and_missing_data(self):
"""Test compositor with day portion with alpha_band when SZA data is not provided and there is missing data."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="day_only", include_alpha=True)
res = comp((self.data_b,))
res = res.compute()
expected_l_channel = np.array([[np.nan, 0.], [0.5, 1.]], dtype=np.float32)
expected_alpha = np.array([[np.nan, 1.], [1., 1.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected_l_channel)
np.testing.assert_allclose(res.values[-1], expected_alpha)
[docs]
def test_day_only_area_without_alpha(self):
"""Test compositor with day portion without alpha_band when SZA data is not provided."""
from satpy.composites import DayNightCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = DayNightCompositor(name="dn_test", day_night="day_only", include_alpha=False)
res = comp((self.data_a,))
res = res.compute()
expected = np.array([[0., 0.33164983], [0.66835017, 1.]], dtype=np.float32)
assert res.dtype == np.float32
np.testing.assert_allclose(res.values[0], expected)
assert "A" not in res.bands
[docs]
class TestFillingCompositor(unittest.TestCase):
"""Test case for the filling compositor."""
[docs]
def test_fill(self):
"""Test filling."""
from satpy.composites import FillingCompositor
comp = FillingCompositor(name="fill_test")
filler = xr.DataArray(np.array([1, 2, 3, 4, 3, 2, 1]))
red = xr.DataArray(np.array([1, 2, 3, np.nan, 3, 2, 1]))
green = xr.DataArray(np.array([np.nan, 2, 3, 4, 3, 2, np.nan]))
blue = xr.DataArray(np.array([4, 3, 2, 1, 2, 3, 4]))
res = comp([filler, red, green, blue])
np.testing.assert_allclose(res.sel(bands="R").data, filler.data)
np.testing.assert_allclose(res.sel(bands="G").data, filler.data)
np.testing.assert_allclose(res.sel(bands="B").data, blue.data)
[docs]
class TestMultiFiller(unittest.TestCase):
"""Test case for the MultiFiller compositor."""
[docs]
def test_fill(self):
"""Test filling."""
from satpy.composites import MultiFiller
comp = MultiFiller(name="fill_test")
attrs = {"units": "K"}
a = xr.DataArray(np.array([1, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan]), attrs=attrs.copy())
b = xr.DataArray(np.array([np.nan, 2, 3, np.nan, np.nan, np.nan, np.nan]), attrs=attrs.copy())
c = xr.DataArray(np.array([np.nan, 22, 3, np.nan, np.nan, np.nan, 7]), attrs=attrs.copy())
d = xr.DataArray(np.array([np.nan, np.nan, np.nan, np.nan, np.nan, 6, np.nan]), attrs=attrs.copy())
e = xr.DataArray(np.array([np.nan, np.nan, np.nan, np.nan, 5, np.nan, np.nan]), attrs=attrs.copy())
expected = xr.DataArray(np.array([1, 2, 3, np.nan, 5, 6, 7]))
res = comp([a, b, c], optional_datasets=[d, e])
np.testing.assert_allclose(res.data, expected.data)
assert "units" in res.attrs
assert res.attrs["units"] == "K"
[docs]
class TestLuminanceSharpeningCompositor(unittest.TestCase):
"""Test luminance sharpening compositor."""
[docs]
def test_compositor(self):
"""Test luminance sharpening compositor."""
from satpy.composites import LuminanceSharpeningCompositor
comp = LuminanceSharpeningCompositor(name="test")
# Three shades of grey
rgb_arr = np.array([1, 50, 100, 200, 1, 50, 100, 200, 1, 50, 100, 200])
rgb = xr.DataArray(rgb_arr.reshape((3, 2, 2)),
dims=["bands", "y", "x"], coords={"bands": ["R", "G", "B"]})
# 100 % luminance -> all result values ~1.0
lum = xr.DataArray(np.array([[100., 100.], [100., 100.]]),
dims=["y", "x"])
res = comp([lum, rgb])
np.testing.assert_allclose(res.data, 1., atol=1e-9)
# 50 % luminance, all result values ~0.5
lum = xr.DataArray(np.array([[50., 50.], [50., 50.]]),
dims=["y", "x"])
res = comp([lum, rgb])
np.testing.assert_allclose(res.data, 0.5, atol=1e-9)
# 30 % luminance, all result values ~0.3
lum = xr.DataArray(np.array([[30., 30.], [30., 30.]]),
dims=["y", "x"])
res = comp([lum, rgb])
np.testing.assert_allclose(res.data, 0.3, atol=1e-9)
# 0 % luminance, all values ~0.0
lum = xr.DataArray(np.array([[0., 0.], [0., 0.]]),
dims=["y", "x"])
res = comp([lum, rgb])
np.testing.assert_allclose(res.data, 0.0, atol=1e-9)
[docs]
class TestSandwichCompositor:
"""Test sandwich compositor."""
# Test RGB and RGBA
[docs]
@pytest.mark.parametrize(
("input_shape", "bands"),
[
((3, 2, 2), ["R", "G", "B"]),
((4, 2, 2), ["R", "G", "B", "A"])
]
)
@mock.patch("satpy.composites.enhance2dataset")
def test_compositor(self, e2d, input_shape, bands):
"""Test luminance sharpening compositor."""
from satpy.composites import SandwichCompositor
rgb_arr = da.from_array(np.random.random(input_shape), chunks=2)
rgb = xr.DataArray(rgb_arr, dims=["bands", "y", "x"],
coords={"bands": bands})
lum_arr = da.from_array(100 * np.random.random((2, 2)), chunks=2)
lum = xr.DataArray(lum_arr, dims=["y", "x"])
# Make enhance2dataset return unmodified dataset
e2d.return_value = rgb
comp = SandwichCompositor(name="test")
res = comp([lum, rgb])
for band in rgb:
if band.bands != "A":
# Check compositor has modified this band
np.testing.assert_allclose(res.loc[band.bands].to_numpy(),
band.to_numpy() * lum_arr / 100.)
else:
# Check Alpha band remains intact
np.testing.assert_allclose(res.loc[band.bands].to_numpy(),
band.to_numpy())
# make sure the compositor doesn't modify the input data
np.testing.assert_allclose(lum.values, lum_arr.compute())
[docs]
class TestInlineComposites(unittest.TestCase):
"""Test inline composites."""
[docs]
def test_inline_composites(self):
"""Test that inline composites are working."""
from satpy.composites.config_loader import load_compositor_configs_for_sensors
comps = load_compositor_configs_for_sensors(["visir"])[0]
# Check that "fog" product has all its prerequisites defined
keys = comps["visir"].keys()
fog = [comps["visir"][dsid] for dsid in keys if "fog" == dsid["name"]][0]
assert fog.attrs["prerequisites"][0]["name"] == "_fog_dep_0"
assert fog.attrs["prerequisites"][1]["name"] == "_fog_dep_1"
assert fog.attrs["prerequisites"][2] == 10.8
# Check that the sub-composite dependencies use wavelengths
# (numeric values)
keys = comps["visir"].keys()
fog_dep_ids = [dsid for dsid in keys if "fog_dep" in dsid["name"]]
assert comps["visir"][fog_dep_ids[0]].attrs["prerequisites"] == [12.0, 10.8]
assert comps["visir"][fog_dep_ids[1]].attrs["prerequisites"] == [10.8, 8.7]
# Check the same for SEVIRI and verify channel names are used
# in the sub-composite dependencies instead of wavelengths
comps = load_compositor_configs_for_sensors(["seviri"])[0]
keys = comps["seviri"].keys()
fog_dep_ids = [dsid for dsid in keys if "fog_dep" in dsid["name"]]
assert comps["seviri"][fog_dep_ids[0]].attrs["prerequisites"] == ["IR_120", "IR_108"]
assert comps["seviri"][fog_dep_ids[1]].attrs["prerequisites"] == ["IR_108", "IR_087"]
[docs]
class TestColormapCompositor(unittest.TestCase):
"""Test the ColormapCompositor."""
[docs]
def setUp(self):
"""Set up the test case."""
from satpy.composites import ColormapCompositor
self.colormap_compositor = ColormapCompositor("test_cmap_compositor")
[docs]
def test_build_colormap_with_int_data_and_without_meanings(self):
"""Test colormap building."""
palette = np.array([[0, 0, 0], [127, 127, 127], [255, 255, 255]])
colormap, squeezed_palette = self.colormap_compositor.build_colormap(palette, np.uint8, {})
assert np.allclose(colormap.values, [0, 1])
assert np.allclose(squeezed_palette, palette / 255.0)
[docs]
def test_build_colormap_with_int_data_and_with_meanings(self):
"""Test colormap building."""
palette = xr.DataArray(np.array([[0, 0, 0], [127, 127, 127], [255, 255, 255]]),
dims=["value", "band"])
palette.attrs["palette_meanings"] = [2, 3, 4]
colormap, squeezed_palette = self.colormap_compositor.build_colormap(palette, np.uint8, {})
assert np.allclose(colormap.values, [2, 3, 4])
assert np.allclose(squeezed_palette, palette / 255.0)
[docs]
class TestPaletteCompositor(unittest.TestCase):
"""Test the PaletteCompositor."""
[docs]
def test_call(self):
"""Test palette compositing."""
from satpy.composites import PaletteCompositor
cmap_comp = PaletteCompositor("test_cmap_compositor")
palette = xr.DataArray(np.array([[0, 0, 0], [127, 127, 127], [255, 255, 255]]),
dims=["value", "band"])
palette.attrs["palette_meanings"] = [2, 3, 4]
data = xr.DataArray(da.from_array(np.array([[4, 3, 2], [2, 3, 4]], dtype=np.uint8)), dims=["y", "x"])
res = cmap_comp([data, palette])
exp = np.array([[[1., 0.498039, 0.],
[0., 0.498039, 1.]],
[[1., 0.498039, 0.],
[0., 0.498039, 1.]],
[[1., 0.498039, 0.],
[0., 0.498039, 1.]]])
assert np.allclose(res, exp)
[docs]
class TestColorizeCompositor(unittest.TestCase):
"""Test the ColorizeCompositor."""
[docs]
def test_colorize_no_fill(self):
"""Test colorizing."""
from satpy.composites import ColorizeCompositor
colormap_composite = ColorizeCompositor("test_color_compositor")
palette = xr.DataArray(np.array([[0, 0, 0], [127, 127, 127], [255, 255, 255]]),
dims=["value", "band"])
palette.attrs["palette_meanings"] = [2, 3, 4]
data = xr.DataArray(np.array([[4, 3, 2],
[2, 3, 4]],
dtype=np.uint8),
dims=["y", "x"])
res = colormap_composite([data, palette])
exp = np.array([[[1., 0.498039, 0.],
[0., 0.498039, 1.]],
[[1., 0.498039, 0.],
[0., 0.498039, 1.]],
[[1., 0.498039, 0.],
[0., 0.498039, 1.]]])
assert np.allclose(res, exp, atol=0.0001)
[docs]
def test_colorize_with_interpolation(self):
"""Test colorizing with interpolation."""
from satpy.composites import ColorizeCompositor
colormap_composite = ColorizeCompositor("test_color_compositor")
palette = xr.DataArray(np.array([[0, 0, 0], [127, 127, 127], [255, 255, 255]]),
dims=["value", "band"])
palette.attrs["palette_meanings"] = [2, 3, 4]
data = xr.DataArray(da.from_array(np.array([[4, 3, 2.5],
[2, 3.2, 4]])),
dims=["y", "x"],
attrs={"valid_range": np.array([2, 4])})
res = colormap_composite([data, palette])
exp = np.array([[[1.0, 0.498039, 0.246575],
[0., 0.59309977, 1.0]],
[[1.0, 0.49803924, 0.24657543],
[0., 0.59309983, 1.0]],
[[1.0, 0.4980392, 0.24657541],
[0., 0.59309978, 1.0]]])
np.testing.assert_allclose(res, exp, atol=1e-4)
[docs]
class TestCloudCompositorWithoutCloudfree:
"""Test the CloudCompositorWithoutCloudfree."""
[docs]
def setup_method(self):
"""Set up the test case."""
from satpy.composites.cloud_products import CloudCompositorWithoutCloudfree
self.colormap_composite = CloudCompositorWithoutCloudfree("test_cmap_compositor")
self.exp = np.array([[4, 3, 2], [2, 3, np.nan], [8, 7, 655350]])
self.exp_bad_oc = np.array([[4, 3, 2],
[2, np.nan, 4],
[np.nan, 7, 255]])
[docs]
def test_call_numpy_with_invalid_value_in_status(self):
"""Test the CloudCompositorWithoutCloudfree composite generation."""
status = xr.DataArray(np.array([[0, 0, 0], [0, 0, 65535], [0, 0, 1]]), dims=["y", "x"],
attrs={"_FillValue": 65535})
data = xr.DataArray(np.array([[4, 3, 2], [2, 3, np.nan], [8, 7, np.nan]], dtype=np.float32),
dims=["y", "x"],
attrs={"_FillValue": 65535,
"scaled_FillValue": 655350})
res = self.colormap_composite([data, status])
np.testing.assert_allclose(res, self.exp, atol=1e-4)
[docs]
def test_call_dask_with_invalid_value_in_status(self):
"""Test the CloudCompositorWithoutCloudfree composite generation."""
status = xr.DataArray(da.from_array(np.array([[0, 0, 0], [0, 0, 65535], [0, 0, 1]])), dims=["y", "x"],
attrs={"_FillValue": 65535})
data = xr.DataArray(da.from_array(np.array([[4, 3, 2], [2, 3, np.nan], [8, 7, np.nan]], dtype=np.float32)),
dims=["y", "x"],
attrs={"_FillValue": 99,
"scaled_FillValue": 655350})
res = self.colormap_composite([data, status])
np.testing.assert_allclose(res, self.exp, atol=1e-4)
[docs]
def test_call_bad_optical_conditions(self):
"""Test the CloudCompositorWithoutCloudfree composite generation."""
status = xr.DataArray(da.from_array(np.array([[0, 0, 0], [3, 3, 3], [0, 0, 1]])), dims=["y", "x"],
attrs={"_FillValue": 65535,
"flag_meanings": "bad_optical_conditions"})
data = xr.DataArray(np.array([[4, 3, 2], [2, 255, 4], [255, 7, 255]], dtype=np.uint8),
dims=["y", "x"],
name="cmic_cre",
attrs={"_FillValue": 255,
"scaled_FillValue": 255})
res = self.colormap_composite([data, status])
np.testing.assert_allclose(res, self.exp_bad_oc, atol=1e-4)
[docs]
def test_bad_indata(self):
"""Test the CloudCompositorWithoutCloudfree composite generation without status."""
data = xr.DataArray(np.array([[4, 3, 2], [2, 3, 4], [255, 7, 255]], dtype=np.uint8),
dims=["y", "x"],
attrs={"_FillValue": 255,
"scaled_FillValue": 255})
np.testing.assert_raises(ValueError, self.colormap_composite, [data])
[docs]
class TestCloudCompositorCommonMask:
"""Test the CloudCompositorCommonMask."""
[docs]
def setup_method(self):
"""Set up the test case."""
from satpy.composites.cloud_products import CloudCompositorCommonMask
self.exp_a = np.array([[4, 3, 2],
[2, 3, 655350],
[np.nan, np.nan, np.nan]])
self.exp_b = np.array([[4, 3, 2],
[2, 3, 255],
[np.nan, np.nan, np.nan]])
self.colormap_composite = CloudCompositorCommonMask("test_cmap_compositor")
[docs]
def test_call_numpy(self):
"""Test the CloudCompositorCommonMask with numpy."""
mask = xr.DataArray(np.array([[0, 0, 0], [1, 1, 1], [255, 255, 255]]), dims=["y", "x"],
attrs={"_FillValue": 255})
data = xr.DataArray(np.array([[4, 3, 2], [2, 3, np.nan], [np.nan, np.nan, np.nan]], dtype=np.float32),
dims=["y", "x"],
attrs={"_FillValue": 65535,
"scaled_FillValue": 655350})
res = self.colormap_composite([data, mask])
np.testing.assert_allclose(res, self.exp_a, atol=1e-4)
[docs]
def test_call_dask(self):
"""Test the CloudCompositorCommonMask with dask."""
mask = xr.DataArray(da.from_array(np.array([[0, 0, 0], [1, 1, 1], [255, 255, 255]])), dims=["y", "x"],
attrs={"_FillValue": 255})
data = xr.DataArray(da.from_array(np.array([[4, 3, 2], [2, 3, 255], [255, 255, 255]], dtype=np.int16)),
dims=["y", "x"],
attrs={"_FillValue": 255,
"scaled_FillValue": 255})
res = self.colormap_composite([data, mask])
np.testing.assert_allclose(res, self.exp_b, atol=1e-4)
[docs]
def test_bad_call(self):
"""Test the CloudCompositorCommonMask without mask."""
data = xr.DataArray(np.array([[4, 3, 2], [2, 3, 255], [255, 255, 255]], dtype=np.int16),
dims=["y", "x"],
attrs={"_FillValue": 255,
"scaled_FillValue": 255})
np.testing.assert_raises(ValueError, self.colormap_composite, [data])
[docs]
class TestPrecipCloudsCompositor(unittest.TestCase):
"""Test the PrecipClouds compositor."""
[docs]
def test_call(self):
"""Test the precip composite generation."""
from satpy.composites.cloud_products import PrecipCloudsRGB
colormap_compositor = PrecipCloudsRGB("test_precip_compositor")
data_light = xr.DataArray(np.array([[80, 70, 60, 0], [20, 30, 40, 255]], dtype=np.uint8),
dims=["y", "x"], attrs={"_FillValue": 255})
data_moderate = xr.DataArray(np.array([[60, 50, 40, 0], [20, 30, 40, 255]], dtype=np.uint8),
dims=["y", "x"], attrs={"_FillValue": 255})
data_intense = xr.DataArray(np.array([[40, 30, 20, 0], [20, 30, 40, 255]], dtype=np.uint8),
dims=["y", "x"], attrs={"_FillValue": 255})
data_flags = xr.DataArray(np.array([[0, 0, 4, 0], [0, 0, 0, 0]], dtype=np.uint8),
dims=["y", "x"])
res = colormap_compositor([data_light, data_moderate, data_intense, data_flags])
exp = np.array([[[0.24313725, 0.18235294, 0.12156863, np.nan],
[0.12156863, 0.18235294, 0.24313725, np.nan]],
[[0.62184874, 0.51820728, 0.41456583, np.nan],
[0.20728291, 0.31092437, 0.41456583, np.nan]],
[[0.82913165, 0.7254902, 0.62184874, np.nan],
[0.20728291, 0.31092437, 0.41456583, np.nan]]])
np.testing.assert_allclose(res, exp)
[docs]
class TestHighCloudCompositor:
"""Test HighCloudCompositor."""
[docs]
def setup_method(self):
"""Create test data."""
from pyresample.geometry import create_area_def
area = create_area_def(area_id="test", projection={"proj": "latlong"},
center=(0, 45), width=3, height=3, resolution=35)
self.dtype = np.float32
self.data = xr.DataArray(
da.from_array(np.array([[200, 250, 300], [200, 250, 300], [200, 250, 300]], dtype=self.dtype)),
dims=("y", "x"), coords={"y": [0, 1, 2], "x": [0, 1, 2]},
attrs={"area": area}
)
[docs]
def test_high_cloud_compositor(self):
"""Test general default functionality of compositor."""
from satpy.composites import HighCloudCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = HighCloudCompositor(name="test")
res = comp([self.data])
assert isinstance(res, xr.DataArray)
assert isinstance(res.data, da.Array)
expexted_alpha = np.array([[1.0, 0.7142857, 0.0], [1.0, 0.625, 0.0], [1.0, 0.5555555, 0.0]])
expected = np.stack([self.data, expexted_alpha])
np.testing.assert_almost_equal(res.values, expected)
[docs]
def test_high_cloud_compositor_multiple_calls(self):
"""Test that the modified init variables are reset properly when calling the compositor multiple times."""
from satpy.composites import HighCloudCompositor
comp = HighCloudCompositor(name="test")
res = comp([self.data])
res2 = comp([self.data])
np.testing.assert_equal(res.values, res2.values)
[docs]
def test_high_cloud_compositor_dtype(self):
"""Test that the datatype is not altered by the compositor."""
from satpy.composites import HighCloudCompositor
comp = HighCloudCompositor(name="test")
res = comp([self.data])
assert res.data.dtype == self.dtype
[docs]
def test_high_cloud_compositor_validity_checks(self):
"""Test that errors are raised for invalid input data and settings."""
from satpy.composites import HighCloudCompositor
with pytest.raises(ValueError, match="Expected 2 `transition_min_limits` values, got 1"):
_ = HighCloudCompositor("test", transition_min_limits=(210., ))
with pytest.raises(ValueError, match="Expected 2 `latitude_min_limits` values, got 3"):
_ = HighCloudCompositor("test", latitude_min_limits=(20., 40., 60.))
with pytest.raises(ValueError, match="Expected `transition_max` to be of type float, "
"is of type <class 'tuple'>"):
_ = HighCloudCompositor("test", transition_max=(250., 300.))
comp = HighCloudCompositor("test")
with pytest.raises(ValueError, match="Expected 1 dataset, got 2"):
_ = comp([self.data, self.data])
[docs]
class TestLowCloudCompositor:
"""Test LowCloudCompositor."""
[docs]
def setup_method(self):
"""Create test data."""
self.dtype = np.float32
self.btd = xr.DataArray(
da.from_array(np.array([[0.0, 1.0, 10.0], [0.0, 1.0, 10.0], [0.0, 1.0, 10.0]], dtype=self.dtype)),
dims=("y", "x"), coords={"y": [0, 1, 2], "x": [0, 1, 2]}
)
self.bt_win = xr.DataArray(
da.from_array(np.array([[250, 250, 250], [250, 250, 250], [150, 150, 150]], dtype=self.dtype)),
dims=("y", "x"), coords={"y": [0, 1, 2], "x": [0, 1, 2]}
)
self.lsm = xr.DataArray(
da.from_array(np.array([[0., 0., 0.], [1., 1., 1.], [0., 1., 0.]], dtype=self.dtype)),
dims=("y", "x"), coords={"y": [0, 1, 2], "x": [0, 1, 2]}
)
[docs]
def test_low_cloud_compositor(self):
"""Test general default functionality of compositor."""
from satpy.composites import LowCloudCompositor
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = LowCloudCompositor(name="test")
res = comp([self.btd, self.bt_win, self.lsm])
assert isinstance(res, xr.DataArray)
assert isinstance(res.data, da.Array)
expexted_alpha = np.array([[0.0, 0.25, 1.0], [0.0, 0.25, 1.0], [0.0, 0.0, 0.0]])
expected = np.stack([self.btd, expexted_alpha])
np.testing.assert_equal(res.values, expected)
[docs]
def test_low_cloud_compositor_dtype(self):
"""Test that the datatype is not altered by the compositor."""
from satpy.composites import LowCloudCompositor
comp = LowCloudCompositor(name="test")
res = comp([self.btd, self.bt_win, self.lsm])
assert res.data.dtype == self.dtype
[docs]
def test_low_cloud_compositor_validity_checks(self):
"""Test that errors are raised for invalid input data and settings."""
from satpy.composites import LowCloudCompositor
with pytest.raises(ValueError, match="Expected 2 `range_land` values, got 1"):
_ = LowCloudCompositor("test", range_land=(2.0, ))
with pytest.raises(ValueError, match="Expected 2 `range_water` values, got 1"):
_ = LowCloudCompositor("test", range_water=(2.0,))
comp = LowCloudCompositor("test")
with pytest.raises(ValueError, match="Expected 3 datasets, got 2"):
_ = comp([self.btd, self.lsm])
[docs]
class TestSingleBandCompositor(unittest.TestCase):
"""Test the single-band compositor."""
[docs]
def setUp(self):
"""Create test data."""
from satpy.composites import SingleBandCompositor
self.comp = SingleBandCompositor(name="test")
all_valid = np.ones((2, 2))
self.all_valid = xr.DataArray(all_valid, dims=["y", "x"])
[docs]
def test_call(self):
"""Test calling the compositor."""
# Dataset with extra attributes
all_valid = self.all_valid
all_valid.attrs["sensor"] = "foo"
attrs = {
"foo": "bar",
"resolution": 333,
"units": "K",
"sensor": {"fake_sensor1", "fake_sensor2"},
"calibration": "BT",
"wavelength": 10.8
}
self.comp.attrs["resolution"] = None
res = self.comp([all_valid], **attrs)
# Verify attributes
assert res.attrs.get("sensor") == "foo"
assert "foo" in res.attrs
assert res.attrs.get("foo") == "bar"
assert "units" in res.attrs
assert "calibration" in res.attrs
assert "modifiers" not in res.attrs
assert res.attrs["wavelength"] == 10.8
assert res.attrs["resolution"] == 333
[docs]
class TestCategoricalDataCompositor(unittest.TestCase):
"""Test composiotor for recategorization of categorical data."""
[docs]
def setUp(self):
"""Create test data."""
attrs = {"name": "foo"}
data = xr.DataArray(da.from_array([[2., 1.], [3., 0.]]), attrs=attrs,
dims=("y", "x"), coords={"y": [0, 1], "x": [0, 1]})
self.data = data
[docs]
def test_basic_recategorization(self):
"""Test general functionality of compositor incl. attributes."""
from satpy.composites import CategoricalDataCompositor
lut = [np.nan, 0, 1, 1]
name = "bar"
comp = CategoricalDataCompositor(name=name, lut=lut)
res = comp([self.data])
res = res.compute()
expected = np.array([[1., 0.], [1., np.nan]])
np.testing.assert_equal(res.values, expected)
np.testing.assert_equal(res.attrs["name"], name)
np.testing.assert_equal(res.attrs["composite_lut"], lut)
[docs]
def test_too_many_datasets(self):
"""Test that ValueError is raised if more than one dataset is provided."""
from satpy.composites import CategoricalDataCompositor
lut = [np.nan, 0, 1, 1]
comp = CategoricalDataCompositor(name="foo", lut=lut)
np.testing.assert_raises(ValueError, comp, [self.data, self.data])
[docs]
class TestGenericCompositor(unittest.TestCase):
"""Test generic compositor."""
[docs]
def setUp(self):
"""Create test data."""
from satpy.composites import GenericCompositor
self.comp = GenericCompositor(name="test")
self.comp2 = GenericCompositor(name="test2", common_channel_mask=False)
all_valid = np.ones((1, 2, 2))
self.all_valid = xr.DataArray(all_valid, dims=["bands", "y", "x"])
first_invalid = np.reshape(np.array([np.nan, 1., 1., 1.]), (1, 2, 2))
self.first_invalid = xr.DataArray(first_invalid,
dims=["bands", "y", "x"])
second_invalid = np.reshape(np.array([1., np.nan, 1., 1.]), (1, 2, 2))
self.second_invalid = xr.DataArray(second_invalid,
dims=["bands", "y", "x"])
wrong_shape = np.reshape(np.array([1., 1., 1.]), (1, 3, 1))
self.wrong_shape = xr.DataArray(wrong_shape, dims=["bands", "y", "x"])
[docs]
def test_masking(self):
"""Test masking in generic compositor."""
# Single channel
res = self.comp([self.all_valid])
np.testing.assert_allclose(res.data, 1., atol=1e-9)
# Three channels, one value invalid
res = self.comp([self.all_valid, self.all_valid, self.first_invalid])
correct = np.reshape(np.array([np.nan, 1., 1., 1.]), (2, 2))
for i in range(3):
np.testing.assert_almost_equal(res.data[i, :, :], correct)
# Three channels, two values invalid
res = self.comp([self.all_valid, self.first_invalid, self.second_invalid])
correct = np.reshape(np.array([np.nan, np.nan, 1., 1.]), (2, 2))
for i in range(3):
np.testing.assert_almost_equal(res.data[i, :, :], correct)
[docs]
def test_concat_datasets(self):
"""Test concatenation of datasets."""
from satpy.composites import IncompatibleAreas
res = self.comp._concat_datasets([self.all_valid], "L")
num_bands = len(res.bands)
assert num_bands == 1
assert res.shape[0] == num_bands
assert res.bands[0] == "L"
res = self.comp._concat_datasets([self.all_valid, self.all_valid], "LA")
num_bands = len(res.bands)
assert num_bands == 2
assert res.shape[0] == num_bands
assert res.bands[0] == "L"
assert res.bands[1] == "A"
with pytest.raises(IncompatibleAreas):
self.comp._concat_datasets([self.all_valid, self.wrong_shape], "LA")
[docs]
def test_get_sensors(self):
"""Test getting sensors from the dataset attributes."""
res = self.comp._get_sensors([self.all_valid])
assert res is None
dset1 = self.all_valid
dset1.attrs["sensor"] = "foo"
res = self.comp._get_sensors([dset1])
assert res == "foo"
dset2 = self.first_invalid
dset2.attrs["sensor"] = "bar"
res = self.comp._get_sensors([dset1, dset2])
assert "foo" in res
assert "bar" in res
assert len(res) == 2
assert isinstance(res, set)
[docs]
@mock.patch("satpy.composites.GenericCompositor._get_sensors")
@mock.patch("satpy.composites.combine_metadata")
@mock.patch("satpy.composites.check_times")
@mock.patch("satpy.composites.GenericCompositor.match_data_arrays")
def test_call_with_mock(self, match_data_arrays, check_times, combine_metadata, get_sensors):
"""Test calling generic compositor."""
from satpy.composites import IncompatibleAreas
combine_metadata.return_value = dict()
get_sensors.return_value = "foo"
# One dataset, no mode given
res = self.comp([self.all_valid])
assert res.shape[0] == 1
assert res.attrs["mode"] == "L"
match_data_arrays.assert_not_called()
# This compositor has been initialized without common masking, so the
# masking shouldn't have been called
projectables = [self.all_valid, self.first_invalid, self.second_invalid]
match_data_arrays.return_value = projectables
res = self.comp2(projectables)
match_data_arrays.assert_called_once()
match_data_arrays.reset_mock()
# Dataset for alpha given, so shouldn't be masked
projectables = [self.all_valid, self.all_valid]
match_data_arrays.return_value = projectables
res = self.comp(projectables)
match_data_arrays.assert_called_once()
match_data_arrays.reset_mock()
# When areas are incompatible, masking shouldn't happen
match_data_arrays.side_effect = IncompatibleAreas()
with pytest.raises(IncompatibleAreas):
self.comp([self.all_valid, self.wrong_shape])
match_data_arrays.assert_called_once()
[docs]
def test_call(self):
"""Test calling generic compositor."""
# Multiple datasets with extra attributes
all_valid = self.all_valid
all_valid.attrs["sensor"] = "foo"
attrs = {"foo": "bar", "resolution": 333}
self.comp.attrs["resolution"] = None
res = self.comp([self.all_valid, self.first_invalid], **attrs)
# Verify attributes
assert res.attrs.get("sensor") == "foo"
assert "foo" in res.attrs
assert res.attrs.get("foo") == "bar"
assert "units" not in res.attrs
assert "calibration" not in res.attrs
assert "modifiers" not in res.attrs
assert res.attrs["wavelength"] is None
assert res.attrs["mode"] == "LA"
assert res.attrs["resolution"] == 333
[docs]
def test_deprecation_warning(self):
"""Test deprecation warning for dcprecated composite recipes."""
warning_message = "foo is a deprecated composite. Use composite bar instead."
self.comp.attrs["deprecation_warning"] = warning_message
with pytest.warns(UserWarning, match=warning_message):
self.comp([self.all_valid])
[docs]
class TestAddBands(unittest.TestCase):
"""Test case for the `add_bands` function."""
[docs]
def test_add_bands_l_rgb(self):
"""Test adding bands."""
from satpy.composites import add_bands
# L + RGB -> RGB
data = xr.DataArray(da.ones((1, 3, 3)), dims=("bands", "y", "x"),
coords={"bands": ["L"]})
new_bands = xr.DataArray(da.array(["R", "G", "B"]), dims=("bands"),
coords={"bands": ["R", "G", "B"]})
res = add_bands(data, new_bands)
res_bands = ["R", "G", "B"]
assert res.attrs["mode"] == "".join(res_bands)
np.testing.assert_array_equal(res.bands, res_bands)
np.testing.assert_array_equal(res.coords["bands"], res_bands)
[docs]
def test_add_bands_l_rgba(self):
"""Test adding bands."""
from satpy.composites import add_bands
# L + RGBA -> RGBA
data = xr.DataArray(da.ones((1, 3, 3)), dims=("bands", "y", "x"),
coords={"bands": ["L"]}, attrs={"mode": "L"})
new_bands = xr.DataArray(da.array(["R", "G", "B", "A"]), dims=("bands"),
coords={"bands": ["R", "G", "B", "A"]})
res = add_bands(data, new_bands)
res_bands = ["R", "G", "B", "A"]
assert res.attrs["mode"] == "".join(res_bands)
np.testing.assert_array_equal(res.bands, res_bands)
np.testing.assert_array_equal(res.coords["bands"], res_bands)
[docs]
def test_add_bands_la_rgb(self):
"""Test adding bands."""
from satpy.composites import add_bands
# LA + RGB -> RGBA
data = xr.DataArray(da.ones((2, 3, 3)), dims=("bands", "y", "x"),
coords={"bands": ["L", "A"]}, attrs={"mode": "LA"})
new_bands = xr.DataArray(da.array(["R", "G", "B"]), dims=("bands"),
coords={"bands": ["R", "G", "B"]})
res = add_bands(data, new_bands)
res_bands = ["R", "G", "B", "A"]
assert res.attrs["mode"] == "".join(res_bands)
np.testing.assert_array_equal(res.bands, res_bands)
np.testing.assert_array_equal(res.coords["bands"], res_bands)
[docs]
def test_add_bands_rgb_rbga(self):
"""Test adding bands."""
from satpy.composites import add_bands
# RGB + RGBA -> RGBA
data = xr.DataArray(da.ones((3, 3, 3)), dims=("bands", "y", "x"),
coords={"bands": ["R", "G", "B"]},
attrs={"mode": "RGB"})
new_bands = xr.DataArray(da.array(["R", "G", "B", "A"]), dims=("bands"),
coords={"bands": ["R", "G", "B", "A"]})
res = add_bands(data, new_bands)
res_bands = ["R", "G", "B", "A"]
assert res.attrs["mode"] == "".join(res_bands)
np.testing.assert_array_equal(res.bands, res_bands)
np.testing.assert_array_equal(res.coords["bands"], res_bands)
[docs]
def test_add_bands_p_l(self):
"""Test adding bands."""
from satpy.composites import add_bands
# P(RGBA) + L -> RGBA
data = xr.DataArray(da.ones((1, 3, 3)), dims=("bands", "y", "x"),
coords={"bands": ["P"]},
attrs={"mode": "P"})
new_bands = xr.DataArray(da.array(["L"]), dims=("bands"),
coords={"bands": ["L"]})
with pytest.raises(NotImplementedError):
add_bands(data, new_bands)
[docs]
class TestStaticImageCompositor(unittest.TestCase):
"""Test case for the static compositor."""
[docs]
@mock.patch("satpy.resample.get_area_def")
def test_init(self, get_area_def):
"""Test the initializiation of static compositor."""
from satpy.composites import StaticImageCompositor
# No filename given raises ValueError
with pytest.raises(ValueError, match="StaticImageCompositor needs a .*"):
StaticImageCompositor("name")
# No area defined
comp = StaticImageCompositor("name", filename="/foo.tif")
assert comp._cache_filename == "/foo.tif"
assert comp.area is None
# Area defined
get_area_def.return_value = "bar"
comp = StaticImageCompositor("name", filename="/foo.tif", area="euro4")
assert comp._cache_filename == "/foo.tif"
assert comp.area == "bar"
get_area_def.assert_called_once_with("euro4")
[docs]
@mock.patch("satpy.aux_download.retrieve")
@mock.patch("satpy.aux_download.register_file")
@mock.patch("satpy.Scene")
def test_call(self, Scene, register, retrieve): # noqa
"""Test the static compositing."""
from satpy.composites import StaticImageCompositor
satpy.config.set(data_dir=os.path.join(os.path.sep, "path", "to", "image"))
remote_tif = "http://example.com/foo.tif"
class MockScene(dict):
def load(self, arg):
pass
img = mock.MagicMock()
img.attrs = {}
scn = MockScene()
scn["image"] = img
Scene.return_value = scn
# absolute path to local file
comp = StaticImageCompositor("name", filename="/foo.tif", area="euro4")
res = comp()
Scene.assert_called_once_with(reader="generic_image",
filenames=["/foo.tif"])
register.assert_not_called()
retrieve.assert_not_called()
assert res.attrs["sensor"] is None
assert "modifiers" not in res.attrs
assert "calibration" not in res.attrs
# remote file with local cached version
Scene.reset_mock()
register.return_value = "data_dir/foo.tif"
retrieve.return_value = "data_dir/foo.tif"
comp = StaticImageCompositor("name", url=remote_tif, area="euro4")
res = comp()
Scene.assert_called_once_with(reader="generic_image",
filenames=["data_dir/foo.tif"])
assert res.attrs["sensor"] is None
assert "modifiers" not in res.attrs
assert "calibration" not in res.attrs
# Non-georeferenced image, no area given
img.attrs.pop("area")
comp = StaticImageCompositor("name", filename="/foo.tif")
with pytest.raises(AttributeError):
comp()
# Non-georeferenced image, area given
comp = StaticImageCompositor("name", filename="/foo.tif", area="euro4")
res = comp()
assert res.attrs["area"].area_id == "euro4"
# Filename contains environment variable
os.environ["TEST_IMAGE_PATH"] = "/path/to/image"
comp = StaticImageCompositor("name", filename="${TEST_IMAGE_PATH}/foo.tif", area="euro4")
assert comp._cache_filename == "/path/to/image/foo.tif"
# URL and filename without absolute path
comp = StaticImageCompositor("name", url=remote_tif, filename="bar.tif")
assert comp._url == remote_tif
assert comp._cache_filename == "bar.tif"
# No URL, filename without absolute path, use default data_dir from config
with mock.patch("os.path.exists") as exists:
exists.return_value = True
comp = StaticImageCompositor("name", filename="foo.tif")
assert comp._url is None
assert comp._cache_filename == os.path.join(os.path.sep, "path", "to", "image", "foo.tif")
[docs]
def _enhance2dataset(dataset, convert_p=False):
"""Mock the enhance2dataset to return the original data."""
return dataset
[docs]
class TestBackgroundCompositor:
"""Test case for the background compositor."""
[docs]
@classmethod
def setup_class(cls):
"""Create shared input data arrays."""
foreground_data = {
"L": np.array([[[1., 0.5], [0., np.nan]]]),
"LA": np.array([[[1., 0.5], [0., np.nan]], [[0.5, 0.5], [0.5, 0.5]]]),
"RGB": np.array([
[[1., 0.5], [0., np.nan]],
[[1., 0.5], [0., np.nan]],
[[1., 0.5], [0., np.nan]]]),
"RGBA": np.array([
[[1.0, 0.5], [0.0, np.nan]],
[[1.0, 0.5], [0.0, np.nan]],
[[1.0, 0.5], [0.0, np.nan]],
[[0.5, 0.5], [0.5, 0.5]]]),
}
cls.foreground_data = foreground_data
[docs]
@mock.patch("satpy.composites.enhance2dataset", _enhance2dataset)
@pytest.mark.parametrize(
("foreground_bands", "background_bands", "exp_bands", "exp_result"),
[
("L", "L", "L", np.array([[1.0, 0.5], [0.0, 1.0]])),
("LA", "LA", "L", np.array([[1.0, 0.75], [0.5, 1.0]])),
("RGB", "RGB", "RGB", np.array([
[[1., 0.5], [0., 1.]],
[[1., 0.5], [0., 1.]],
[[1., 0.5], [0., 1.]]])),
("RGBA", "RGBA", "RGB", np.array([
[[1., 0.75], [0.5, 1.]],
[[1., 0.75], [0.5, 1.]],
[[1., 0.75], [0.5, 1.]]])),
("RGBA", "RGB", "RGB", np.array([
[[1., 0.75], [0.5, 1.]],
[[1., 0.75], [0.5, 1.]],
[[1., 0.75], [0.5, 1.]]])),
]
)
def test_call(self, foreground_bands, background_bands, exp_bands, exp_result):
"""Test the background compositing."""
from satpy.composites import BackgroundCompositor
comp = BackgroundCompositor("name")
# L mode images
foreground_data = self.foreground_data[foreground_bands]
attrs = {"mode": foreground_bands, "area": "foo"}
foreground = xr.DataArray(da.from_array(foreground_data),
dims=("bands", "y", "x"),
coords={"bands": [c for c in attrs["mode"]]},
attrs=attrs)
attrs = {"mode": background_bands, "area": "foo"}
background = xr.DataArray(da.ones((len(background_bands), 2, 2)), dims=("bands", "y", "x"),
coords={"bands": [c for c in attrs["mode"]]},
attrs=attrs)
res = comp([foreground, background])
assert res.attrs["area"] == "foo"
np.testing.assert_allclose(res, exp_result)
assert res.attrs["mode"] == exp_bands
[docs]
@mock.patch("satpy.composites.enhance2dataset", _enhance2dataset)
def test_multiple_sensors(self):
"""Test the background compositing from multiple sensor data."""
from satpy.composites import BackgroundCompositor
comp = BackgroundCompositor("name")
# L mode images
attrs = {"mode": "L", "area": "foo"}
foreground_data = self.foreground_data["L"]
foreground = xr.DataArray(da.from_array(foreground_data),
dims=("bands", "y", "x"),
coords={"bands": [c for c in attrs["mode"]]},
attrs=attrs.copy())
foreground.attrs["sensor"] = "abi"
background = xr.DataArray(da.ones((1, 2, 2)), dims=("bands", "y", "x"),
coords={"bands": [c for c in attrs["mode"]]},
attrs=attrs.copy())
background.attrs["sensor"] = "glm"
res = comp([foreground, background])
assert res.attrs["area"] == "foo"
np.testing.assert_allclose(res, np.array([[1., 0.5], [0., 1.]]))
assert res.attrs["mode"] == "L"
assert res.attrs["sensor"] == {"abi", "glm"}
[docs]
class TestMaskingCompositor:
"""Test case for the simple masking compositor."""
[docs]
@pytest.fixture()
def conditions_v1(self):
"""Masking conditions with string values."""
return [{"method": "equal",
"value": "Cloud-free_land",
"transparency": 100},
{"method": "equal",
"value": "Cloud-free_sea",
"transparency": 50}]
[docs]
@pytest.fixture()
def conditions_v2(self):
"""Masking conditions with numerical values."""
return [{"method": "equal",
"value": 1,
"transparency": 100},
{"method": "equal",
"value": 2,
"transparency": 50}]
[docs]
@pytest.fixture()
def test_data(self):
"""Test data to use with masking compositors."""
return xr.DataArray(da.random.random((3, 3)), dims=["y", "x"])
[docs]
@pytest.fixture()
def test_ct_data(self):
"""Test 2D CT data array."""
flag_meanings = ["Cloud-free_land", "Cloud-free_sea"]
flag_values = da.array([1, 2])
ct_data = da.array([[1, 2, 2],
[2, 1, 2],
[2, 2, 1]])
ct_data = xr.DataArray(ct_data, dims=["y", "x"])
ct_data.attrs["flag_meanings"] = flag_meanings
ct_data.attrs["flag_values"] = flag_values
return ct_data
[docs]
@pytest.fixture()
def test_ct_data_v3(self, test_ct_data):
"""Set ct data to NaN where it originally is 1."""
return test_ct_data.where(test_ct_data == 1)
[docs]
@pytest.fixture()
def reference_data(self, test_data, test_ct_data):
"""Get reference data to use in masking compositor tests."""
# The data are set to NaN where ct is `1`
return test_data.where(test_ct_data > 1)
[docs]
@pytest.fixture()
def reference_alpha(self):
"""Get reference alpha to use in masking compositor tests."""
ref_alpha = da.array([[0, 0.5, 0.5],
[0.5, 0, 0.5],
[0.5, 0.5, 0]])
return xr.DataArray(ref_alpha, dims=["y", "x"])
[docs]
def test_init(self):
"""Test the initializiation of compositor."""
from satpy.composites import MaskingCompositor
# No transparency or conditions given raises ValueError
with pytest.raises(ValueError, match="Masking conditions not defined."):
_ = MaskingCompositor("name")
# transparency defined
transparency = {0: 100, 1: 50}
conditions = [{"method": "equal", "value": 0, "transparency": 100},
{"method": "equal", "value": 1, "transparency": 50}]
comp = MaskingCompositor("name", transparency=transparency.copy())
assert not hasattr(comp, "transparency")
# Transparency should be converted to conditions
assert comp.conditions == conditions
# conditions defined
comp = MaskingCompositor("name", conditions=conditions.copy())
assert comp.conditions == conditions
[docs]
def test_get_flag_value(self):
"""Test reading flag value from attributes based on a name."""
from satpy.composites import _get_flag_value
flag_values = da.array([1, 2])
mask = da.array([[1, 2, 2],
[2, 1, 2],
[2, 2, 1]])
mask = xr.DataArray(mask, dims=["y", "x"])
flag_meanings = ["Cloud-free_land", "Cloud-free_sea"]
mask.attrs["flag_meanings"] = flag_meanings
mask.attrs["flag_values"] = flag_values
assert _get_flag_value(mask, "Cloud-free_land") == 1
assert _get_flag_value(mask, "Cloud-free_sea") == 2
flag_meanings_str = "Cloud-free_land Cloud-free_sea"
mask.attrs["flag_meanings"] = flag_meanings_str
assert _get_flag_value(mask, "Cloud-free_land") == 1
assert _get_flag_value(mask, "Cloud-free_sea") == 2
[docs]
@pytest.mark.parametrize("mode", ["LA", "RGBA"])
def test_call_numerical_transparency_data(
self, conditions_v1, test_data, test_ct_data, reference_data,
reference_alpha, mode):
"""Test call the compositor with numerical transparency data.
Use parameterisation to test different image modes.
"""
from satpy.composites import MaskingCompositor
from satpy.tests.utils import CustomScheduler
# Test with numerical transparency data
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = MaskingCompositor("name", conditions=conditions_v1,
mode=mode)
res = comp([test_data, test_ct_data])
assert res.mode == mode
for m in mode.rstrip("A"):
np.testing.assert_allclose(res.sel(bands=m), reference_data)
np.testing.assert_allclose(res.sel(bands="A"), reference_alpha)
[docs]
def test_call_named_fields(self, conditions_v2, test_data, test_ct_data,
reference_data, reference_alpha):
"""Test with named fields."""
from satpy.composites import MaskingCompositor
from satpy.tests.utils import CustomScheduler
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = MaskingCompositor("name", conditions=conditions_v2)
res = comp([test_data, test_ct_data])
assert res.mode == "LA"
np.testing.assert_allclose(res.sel(bands="L"), reference_data)
np.testing.assert_allclose(res.sel(bands="A"), reference_alpha)
[docs]
def test_call_named_fields_string(
self, conditions_v2, test_data, test_ct_data, reference_data,
reference_alpha):
"""Test with named fields which are as a string in the mask attributes."""
from satpy.composites import MaskingCompositor
from satpy.tests.utils import CustomScheduler
flag_meanings_str = "Cloud-free_land Cloud-free_sea"
test_ct_data.attrs["flag_meanings"] = flag_meanings_str
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = MaskingCompositor("name", conditions=conditions_v2)
res = comp([test_data, test_ct_data])
assert res.mode == "LA"
np.testing.assert_allclose(res.sel(bands="L"), reference_data)
np.testing.assert_allclose(res.sel(bands="A"), reference_alpha)
[docs]
def test_method_isnan(self, test_data,
test_ct_data, test_ct_data_v3):
"""Test "isnan" as method."""
from satpy.composites import MaskingCompositor
from satpy.tests.utils import CustomScheduler
conditions_v3 = [{"method": "isnan", "transparency": 100}]
# The data are set to NaN where ct is NaN
reference_data_v3 = test_data.where(test_ct_data == 1)
reference_alpha_v3 = da.array([[1., 0., 0.],
[0., 1., 0.],
[0., 0., 1.]])
reference_alpha_v3 = xr.DataArray(reference_alpha_v3, dims=["y", "x"])
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = MaskingCompositor("name", conditions=conditions_v3)
res = comp([test_data, test_ct_data_v3])
assert res.mode == "LA"
np.testing.assert_allclose(res.sel(bands="L"), reference_data_v3)
np.testing.assert_allclose(res.sel(bands="A"), reference_alpha_v3)
[docs]
def test_method_absolute_import(self, test_data, test_ct_data_v3):
"""Test "absolute_import" as method."""
from satpy.composites import MaskingCompositor
from satpy.tests.utils import CustomScheduler
conditions_v4 = [{"method": "absolute_import", "transparency": "satpy.resample"}]
# This should raise AttributeError
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = MaskingCompositor("name", conditions=conditions_v4)
with pytest.raises(AttributeError):
comp([test_data, test_ct_data_v3])
[docs]
def test_rgb_dataset(self, conditions_v1, test_ct_data, reference_alpha):
"""Test RGB dataset."""
from satpy.composites import MaskingCompositor
from satpy.tests.utils import CustomScheduler
# 3D data array
data = xr.DataArray(da.random.random((3, 3, 3)),
dims=["bands", "y", "x"],
coords={"bands": ["R", "G", "B"],
"y": np.arange(3),
"x": np.arange(3)})
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = MaskingCompositor("name", conditions=conditions_v1)
res = comp([data, test_ct_data])
assert res.mode == "RGBA"
np.testing.assert_allclose(res.sel(bands="R"),
data.sel(bands="R").where(test_ct_data > 1))
np.testing.assert_allclose(res.sel(bands="G"),
data.sel(bands="G").where(test_ct_data > 1))
np.testing.assert_allclose(res.sel(bands="B"),
data.sel(bands="B").where(test_ct_data > 1))
np.testing.assert_allclose(res.sel(bands="A"), reference_alpha)
[docs]
def test_rgba_dataset(self, conditions_v2, test_ct_data, reference_alpha):
"""Test RGBA dataset."""
from satpy.composites import MaskingCompositor
from satpy.tests.utils import CustomScheduler
data = xr.DataArray(da.random.random((4, 3, 3)),
dims=["bands", "y", "x"],
coords={"bands": ["R", "G", "B", "A"],
"y": np.arange(3),
"x": np.arange(3)})
with dask.config.set(scheduler=CustomScheduler(max_computes=0)):
comp = MaskingCompositor("name", conditions=conditions_v2)
res = comp([data, test_ct_data])
assert res.mode == "RGBA"
np.testing.assert_allclose(res.sel(bands="R"),
data.sel(bands="R").where(test_ct_data > 1))
np.testing.assert_allclose(res.sel(bands="G"),
data.sel(bands="G").where(test_ct_data > 1))
np.testing.assert_allclose(res.sel(bands="B"),
data.sel(bands="B").where(test_ct_data > 1))
# The compositor should drop the original alpha band
np.testing.assert_allclose(res.sel(bands="A"), reference_alpha)
[docs]
def test_incorrect_method(self, test_data, test_ct_data):
"""Test incorrect method."""
from satpy.composites import MaskingCompositor
conditions = [{"method": "foo", "value": 0, "transparency": 100}]
comp = MaskingCompositor("name", conditions=conditions)
with pytest.raises(AttributeError):
comp([test_data, test_ct_data])
# Test with too few projectables.
with pytest.raises(ValueError, match="Expected 2 datasets, got 1"):
comp([test_data])
[docs]
def test_incorrect_mode(self, conditions_v1):
"""Test initiating with unsupported mode."""
from satpy.composites import MaskingCompositor
# Incorrect mode raises ValueError
with pytest.raises(ValueError, match="Invalid mode YCbCrA. Supported modes: .*"):
MaskingCompositor("name", conditions=conditions_v1,
mode="YCbCrA")
[docs]
class TestNaturalEnhCompositor(unittest.TestCase):
"""Test NaturalEnh compositor."""
[docs]
def setUp(self):
"""Create channel data and set channel weights."""
self.ch1 = xr.DataArray([1.0])
self.ch2 = xr.DataArray([2.0])
self.ch3 = xr.DataArray([3.0])
self.ch16_w = 2.0
self.ch08_w = 3.0
self.ch06_w = 4.0
[docs]
@mock.patch("satpy.composites.NaturalEnh.__repr__")
@mock.patch("satpy.composites.NaturalEnh.match_data_arrays")
def test_natural_enh(self, match_data_arrays, repr_):
"""Test NaturalEnh compositor."""
from satpy.composites import NaturalEnh
repr_.return_value = ""
projectables = [self.ch1, self.ch2, self.ch3]
def temp_func(*args):
return args[0]
match_data_arrays.side_effect = temp_func
comp = NaturalEnh("foo", ch16_w=self.ch16_w, ch08_w=self.ch08_w,
ch06_w=self.ch06_w)
assert comp.ch16_w == self.ch16_w
assert comp.ch08_w == self.ch08_w
assert comp.ch06_w == self.ch06_w
res = comp(projectables)
assert mock.call(projectables) in match_data_arrays.mock_calls
correct = (self.ch16_w * projectables[0] +
self.ch08_w * projectables[1] +
self.ch06_w * projectables[2])
assert res[0] == correct
assert res[1] == projectables[1]
assert res[2] == projectables[2]
[docs]
class TestEnhance2Dataset(unittest.TestCase):
"""Test the enhance2dataset utility."""
[docs]
@mock.patch("satpy.composites.get_enhanced_image")
def test_enhance_p_to_rgb(self, get_enhanced_image):
"""Test enhancing a paletted dataset in RGB mode."""
from trollimage.xrimage import XRImage
img = XRImage(xr.DataArray(np.ones((1, 20, 20)) * 2, dims=("bands", "y", "x"), coords={"bands": ["P"]}))
img.palette = ((0, 0, 0), (4, 4, 4), (8, 8, 8))
get_enhanced_image.return_value = img
from satpy.composites import enhance2dataset
dataset = xr.DataArray(np.ones((1, 20, 20)))
res = enhance2dataset(dataset, convert_p=True)
assert res.attrs["mode"] == "RGB"
[docs]
@mock.patch("satpy.composites.get_enhanced_image")
def test_enhance_p_to_rgba(self, get_enhanced_image):
"""Test enhancing a paletted dataset in RGBA mode."""
from trollimage.xrimage import XRImage
img = XRImage(xr.DataArray(np.ones((1, 20, 20)) * 2, dims=("bands", "y", "x"), coords={"bands": ["P"]}))
img.palette = ((0, 0, 0, 255), (4, 4, 4, 255), (8, 8, 8, 255))
get_enhanced_image.return_value = img
from satpy.composites import enhance2dataset
dataset = xr.DataArray(np.ones((1, 20, 20)))
res = enhance2dataset(dataset, convert_p=True)
assert res.attrs["mode"] == "RGBA"
[docs]
@mock.patch("satpy.composites.get_enhanced_image")
def test_enhance_p(self, get_enhanced_image):
"""Test enhancing a paletted dataset in P mode."""
from trollimage.xrimage import XRImage
img = XRImage(xr.DataArray(np.ones((1, 20, 20)) * 2, dims=("bands", "y", "x"), coords={"bands": ["P"]}))
img.palette = ((0, 0, 0, 255), (4, 4, 4, 255), (8, 8, 8, 255))
get_enhanced_image.return_value = img
from satpy.composites import enhance2dataset
dataset = xr.DataArray(np.ones((1, 20, 20)))
res = enhance2dataset(dataset)
assert res.attrs["mode"] == "P"
assert res.max().values == 2
[docs]
@mock.patch("satpy.composites.get_enhanced_image")
def test_enhance_l(self, get_enhanced_image):
"""Test enhancing a paletted dataset in P mode."""
from trollimage.xrimage import XRImage
img = XRImage(xr.DataArray(np.ones((1, 20, 20)) * 2, dims=("bands", "y", "x"), coords={"bands": ["L"]}))
get_enhanced_image.return_value = img
from satpy.composites import enhance2dataset
dataset = xr.DataArray(np.ones((1, 20, 20)))
res = enhance2dataset(dataset)
assert res.attrs["mode"] == "L"
assert res.max().values == 1
[docs]
class TestInferMode(unittest.TestCase):
"""Test the infer_mode utility."""
[docs]
def test_bands_coords_is_used(self):
"""Test that the `bands` coord is used."""
from satpy.composites import GenericCompositor
arr = xr.DataArray(np.ones((1, 5, 5)), dims=("bands", "x", "y"), coords={"bands": ["P"]})
assert GenericCompositor.infer_mode(arr) == "P"
arr = xr.DataArray(np.ones((3, 5, 5)), dims=("bands", "x", "y"), coords={"bands": ["Y", "Cb", "Cr"]})
assert GenericCompositor.infer_mode(arr) == "YCbCr"
[docs]
def test_mode_is_used(self):
"""Test that the `mode` attribute is used."""
from satpy.composites import GenericCompositor
arr = xr.DataArray(np.ones((1, 5, 5)), dims=("bands", "x", "y"), attrs={"mode": "P"})
assert GenericCompositor.infer_mode(arr) == "P"
[docs]
def test_band_size_is_used(self):
"""Test that the band size is used."""
from satpy.composites import GenericCompositor
arr = xr.DataArray(np.ones((2, 5, 5)), dims=("bands", "x", "y"))
assert GenericCompositor.infer_mode(arr) == "LA"
[docs]
def test_no_bands_is_l(self):
"""Test that default (no band) is L."""
from satpy.composites import GenericCompositor
arr = xr.DataArray(np.ones((5, 5)), dims=("x", "y"))
assert GenericCompositor.infer_mode(arr) == "L"
[docs]
class TestLongitudeMaskingCompositor(unittest.TestCase):
"""Test case for the LongitudeMaskingCompositor compositor."""
[docs]
def test_masking(self):
"""Test longitude masking."""
from satpy.composites import LongitudeMaskingCompositor
area = mock.MagicMock()
lons = np.array([-180., -100., -50., 0., 50., 100., 180.])
area.get_lonlats = mock.MagicMock(return_value=[lons, []])
a = xr.DataArray(np.array([1, 2, 3, 4, 5, 6, 7]),
attrs={"area": area, "units": "K"})
comp = LongitudeMaskingCompositor(name="test", lon_min=-40., lon_max=120.)
expected = xr.DataArray(np.array([np.nan, np.nan, np.nan, 4, 5, 6, np.nan]))
res = comp([a])
np.testing.assert_allclose(res.data, expected.data)
assert "units" in res.attrs
assert res.attrs["units"] == "K"
comp = LongitudeMaskingCompositor(name="test", lon_min=-40.)
expected = xr.DataArray(np.array([np.nan, np.nan, np.nan, 4, 5, 6, 7]))
res = comp([a])
np.testing.assert_allclose(res.data, expected.data)
comp = LongitudeMaskingCompositor(name="test", lon_max=120.)
expected = xr.DataArray(np.array([1, 2, 3, 4, 5, 6, np.nan]))
res = comp([a])
np.testing.assert_allclose(res.data, expected.data)
comp = LongitudeMaskingCompositor(name="test", lon_min=120., lon_max=-40.)
expected = xr.DataArray(np.array([1, 2, 3, np.nan, np.nan, np.nan, 7]))
res = comp([a])
np.testing.assert_allclose(res.data, expected.data)
[docs]
def test_bad_sensor_yaml_configs(tmp_path):
"""Test composite YAML file with no sensor isn't loaded.
But the bad YAML also shouldn't crash composite configuration loading.
"""
from satpy.composites.config_loader import load_compositor_configs_for_sensors
comp_dir = tmp_path / "composites"
comp_dir.mkdir()
comp_yaml = comp_dir / "fake_sensor.yaml"
with satpy.config.set(config_path=[tmp_path]):
_create_fake_composite_config(comp_yaml)
# no sensor_name in YAML, quietly ignored
comps, _ = load_compositor_configs_for_sensors(["fake_sensor"])
assert "fake_sensor" in comps
assert "fake_composite" not in comps["fake_sensor"]
[docs]
def _create_fake_composite_config(yaml_filename: str):
import yaml
from satpy.composites import StaticImageCompositor
with open(yaml_filename, "w") as comp_file:
yaml.dump({
"composites": {
"fake_composite": {
"compositor": StaticImageCompositor,
"url": "http://example.com/image.png",
},
},
},
comp_file,
)
[docs]
class TestRealisticColors:
"""Test the SEVIRI Realistic Colors compositor."""
[docs]
def test_realistic_colors(self):
"""Test the compositor."""
from satpy.composites import RealisticColors
vis06 = xr.DataArray(da.arange(0, 15, dtype=np.float32).reshape(3, 5), dims=("y", "x"),
attrs={"foo": "foo"})
vis08 = xr.DataArray(da.arange(15, 0, -1, dtype=np.float32).reshape(3, 5), dims=("y", "x"),
attrs={"bar": "bar"})
hrv = xr.DataArray(6 * da.ones((3, 5), dtype=np.float32), dims=("y", "x"),
attrs={"baz": "baz"})
expected_red = np.array([[0.0, 2.733333, 4.9333334, 6.6, 7.733333],
[8.333333, 8.400001, 7.9333334, 7.0, 6.0],
[5.0, 4.0, 3.0, 2.0, 1.0]], dtype=np.float32)
expected_green = np.array([
[15.0, 12.266666, 10.066668, 8.400001, 7.2666664],
[6.6666665, 6.6000004, 7.0666666, 8.0, 9.0],
[10.0, 11.0, 12.0, 13.0, 14.0]], dtype=np.float32)
with dask.config.set(scheduler=CustomScheduler(max_computes=1)):
comp = RealisticColors("Ni!")
res = comp((vis06, vis08, hrv))
arr = res.values
assert res.dtype == np.float32
np.testing.assert_allclose(arr[0, :, :], expected_red)
np.testing.assert_allclose(arr[1, :, :], expected_green)
np.testing.assert_allclose(arr[2, :, :], 3.0)