# Copyright (c) 2020-2023 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/>.
"""Unit tests for the dependency tree class and dependencies."""
import os
import unittest
from satpy.dependency_tree import DependencyTree
from satpy.tests.utils import make_cid, make_dataid
[docs]
class TestDependencyTree(unittest.TestCase):
"""Test the dependency tree.
This is what we are working with::
None (No Data)
+DataID(name='comp19')
+ +DataID(name='ds5', resolution=250, modifiers=('res_change',))
+ + +DataID(name='ds5', resolution=250, modifiers=())
+ + +__EMPTY_LEAF_SENTINEL__ (No Data)
+ +DataID(name='comp13')
+ + +DataID(name='ds5', resolution=250, modifiers=('res_change',))
+ + + +DataID(name='ds5', resolution=250, modifiers=())
+ + + +__EMPTY_LEAF_SENTINEL__ (No Data)
+ +DataID(name='ds2', resolution=250, calibration=<calibration.reflectance>, modifiers=())
"""
[docs]
def setUp(self):
"""Set up the test tree."""
self.dependency_tree = DependencyTree(None, None, None)
composite_1 = make_cid(name="comp19")
dependency_1 = make_dataid(name="ds5", resolution=250, modifiers=("res_change",))
dependency_1_1 = make_dataid(name="ds5", resolution=250, modifiers=tuple())
node_composite_1 = self.dependency_tree.add_leaf(composite_1)
node_dependency_1 = self.dependency_tree.add_leaf(dependency_1, node_composite_1)
self.dependency_tree.add_leaf(dependency_1_1, node_dependency_1)
# ToDo: do we really want then empty node to be at the same level as the unmodified data?
node_dependency_1.add_child(self.dependency_tree.empty_node)
dependency_2 = make_cid(name="comp13")
dependency_2_1 = dependency_1
node_dependency_2 = self.dependency_tree.add_leaf(dependency_2, node_composite_1)
self.dependency_tree.add_leaf(dependency_2_1, node_dependency_2)
# We don't need to add the unmodified dependency a second time.
dependency_3 = make_dataid(name="ds2", resolution=250, calibration="reflectance", modifiers=tuple())
self.dependency_tree.add_leaf(dependency_3, node_composite_1)
[docs]
@staticmethod
def _nodes_equal(node_list1, node_list2):
names1 = [node.name for node in node_list1]
names2 = [node.name for node in node_list2]
return sorted(names1) == sorted(names2)
[docs]
def test_copy_preserves_all_nodes(self):
"""Test that dependency tree copy preserves all nodes."""
new_dependency_tree = self.dependency_tree.copy()
assert self.dependency_tree.empty_node is new_dependency_tree.empty_node
assert self._nodes_equal(self.dependency_tree.leaves(),
new_dependency_tree.leaves())
assert self._nodes_equal(self.dependency_tree.trunk(),
new_dependency_tree.trunk())
# make sure that we can get access to sub-nodes
c13_id = make_cid(name="comp13")
assert self._nodes_equal(self.dependency_tree.trunk(limit_nodes_to=[c13_id]),
new_dependency_tree.trunk(limit_nodes_to=[c13_id]))
[docs]
def test_copy_preserves_unique_empty_node(self):
"""Test that dependency tree copy preserves the uniqueness of the empty node."""
new_dependency_tree = self.dependency_tree.copy()
assert self.dependency_tree.empty_node is new_dependency_tree.empty_node
assert self.dependency_tree._root.children[0].children[0].children[1] is self.dependency_tree.empty_node
assert new_dependency_tree._root.children[0].children[0].children[1] is self.dependency_tree.empty_node
[docs]
def test_new_dependency_tree_preserves_unique_empty_node(self):
"""Test that dependency tree instantiation preserves the uniqueness of the empty node."""
new_dependency_tree = DependencyTree(None, None, None)
assert self.dependency_tree.empty_node is new_dependency_tree.empty_node
[docs]
class TestMissingDependencies(unittest.TestCase):
"""Test the MissingDependencies exception."""
[docs]
def test_new_missing_dependencies(self):
"""Test new MissingDependencies."""
from satpy.node import MissingDependencies
error = MissingDependencies("bla")
assert error.missing_dependencies == "bla"
[docs]
def test_new_missing_dependencies_with_message(self):
"""Test new MissingDependencies with a message."""
from satpy.node import MissingDependencies
error = MissingDependencies("bla", "This is a message")
assert "This is a message" in str(error)
[docs]
class TestMultipleResolutionSameChannelDependency(unittest.TestCase):
"""Test that MODIS situations where the same channel is available at multiple resolution works."""
[docs]
def test_modis_overview_1000m(self):
"""Test a modis overview dependency calculation with resolution fixed to 1000m."""
from satpy import DataQuery
from satpy._config import PACKAGE_CONFIG_PATH
from satpy.composites import GenericCompositor
from satpy.dataset import DatasetDict
from satpy.modifiers.geometry import SunZenithCorrector
from satpy.readers.yaml_reader import FileYAMLReader
config_file = os.path.join(PACKAGE_CONFIG_PATH, "readers", "modis_l1b.yaml")
self.reader_instance = FileYAMLReader.from_config_files(config_file)
overview = {"_satpy_id": make_dataid(name="overview"),
"name": "overview",
"optional_prerequisites": [],
"prerequisites": [DataQuery(name="1", modifiers=("sunz_corrected",)),
DataQuery(name="2", modifiers=("sunz_corrected",)),
DataQuery(name="31")],
"standard_name": "overview"}
compositors = {"modis": DatasetDict()}
compositors["modis"]["overview"] = GenericCompositor(**overview)
modifiers = {"modis": {"sunz_corrected": (SunZenithCorrector,
{"optional_prerequisites": ["solar_zenith_angle"],
"name": "sunz_corrected",
"prerequisites": []})}}
dep_tree = DependencyTree({"modis_l1b": self.reader_instance}, compositors, modifiers)
dep_tree.populate_with_keys({"overview"}, DataQuery(resolution=1000))
for key in dep_tree._all_nodes.keys():
assert key.get("resolution", 1000) == 1000
[docs]
class TestMultipleSensors(unittest.TestCase):
"""Test cases where multiple sensors are available.
This is what we are working with::
None (No Data)
+DataID(name='comp19')
+ +DataID(name='ds5', resolution=250, modifiers=('res_change',))
+ + +DataID(name='ds5', resolution=250, modifiers=())
+ + +__EMPTY_LEAF_SENTINEL__ (No Data)
+ +DataID(name='comp13')
+ + +DataID(name='ds5', resolution=250, modifiers=('res_change',))
+ + + +DataID(name='ds5', resolution=250, modifiers=())
+ + + +__EMPTY_LEAF_SENTINEL__ (No Data)
+ +DataID(name='ds2', resolution=250, calibration=<calibration.reflectance>, modifiers=())
"""
[docs]
def setUp(self):
"""Set up the test tree."""
from satpy.composites import CompositeBase
from satpy.dataset.data_dict import DatasetDict
from satpy.modifiers import ModifierBase
class _FakeCompositor(CompositeBase):
def __init__(self, ret_val, *args, **kwargs):
self.ret_val = ret_val
super().__init__(*args, **kwargs)
def __call__(self, *args, **kwargs):
return self.ret_val
class _FakeModifier(ModifierBase):
def __init__(self, ret_val, *args, **kwargs):
self.ret_val = ret_val
super().__init__(*args, **kwargs)
def __call__(self, *args, **kwargs):
return self.ret_val
comp1_sensor1 = _FakeCompositor(1, "comp1")
comp1_sensor2 = _FakeCompositor(2, "comp1")
# create the dictionary one element at a time to force "incorrect" order
# (sensor2 comes before sensor1, but results should be alphabetical order)
compositors = {}
compositors["sensor2"] = s2_comps = DatasetDict()
compositors["sensor1"] = s1_comps = DatasetDict()
c1_s2_id = make_cid(name="comp1", resolution=1000)
c1_s1_id = make_cid(name="comp1", resolution=500)
s2_comps[c1_s2_id] = comp1_sensor2
s1_comps[c1_s1_id] = comp1_sensor1
modifiers = {}
modifiers["sensor2"] = s2_mods = {}
modifiers["sensor1"] = s1_mods = {}
s2_mods["mod1"] = (_FakeModifier, {"ret_val": 2})
s1_mods["mod1"] = (_FakeModifier, {"ret_val": 1})
self.dependency_tree = DependencyTree({}, compositors, modifiers)
# manually add a leaf so we don't have to mock a reader
ds5 = make_dataid(name="ds5", resolution=250, modifiers=tuple())
self.dependency_tree.add_leaf(ds5)
[docs]
def test_compositor_loaded_sensor_order(self):
"""Test that a compositor is loaded from the first alphabetical sensor."""
self.dependency_tree.populate_with_keys({"comp1"})
comp_nodes = self.dependency_tree.trunk()
assert len(comp_nodes) == 1
assert comp_nodes[0].name["resolution"] == 500
[docs]
def test_modifier_loaded_sensor_order(self):
"""Test that a modifier is loaded from the first alphabetical sensor."""
from satpy import DataQuery
dq = DataQuery(name="ds5", modifiers=("mod1",))
self.dependency_tree.populate_with_keys({dq})
comp_nodes = self.dependency_tree.trunk()
assert len(comp_nodes) == 1
assert comp_nodes[0].data[0].ret_val == 1