#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2019 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/>.
"""Test objects and functions in the satpy.config module."""
from __future__ import annotations
import contextlib
import os
import sys
import unittest
from pathlib import Path
from typing import Callable, Iterator
from unittest import mock
import pkg_resources
import pytest
import satpy
from satpy import DatasetDict
from satpy.composites.config_loader import load_compositor_configs_for_sensors
[docs]class TestBuiltinAreas(unittest.TestCase):
"""Test that the builtin areas are all valid."""
[docs] def test_areas_pyproj(self):
"""Test all areas have valid projections with pyproj."""
import numpy as np
import pyproj
import xarray as xr
from pyresample import parse_area_file
from pyresample.geometry import SwathDefinition
from satpy.resample import get_area_file
lons = np.array([[0, 0.1, 0.2], [0.05, 0.15, 0.25]])
lats = np.array([[0, 0.1, 0.2], [0.05, 0.15, 0.25]])
lons = xr.DataArray(lons)
lats = xr.DataArray(lats)
swath_def = SwathDefinition(lons, lats)
all_areas = parse_area_file(get_area_file())
for area_obj in all_areas:
if hasattr(area_obj, 'freeze'):
try:
area_obj = area_obj.freeze(lonslats=swath_def)
except RuntimeError:
# we didn't provide enough info to freeze, hard to guess
# in a generic test so just skip this area
continue
proj_dict = area_obj.proj_dict
_ = pyproj.Proj(proj_dict)
[docs] def test_areas_rasterio(self):
"""Test all areas have valid projections with rasterio."""
try:
from rasterio.crs import CRS
except ImportError:
return unittest.skip("Missing rasterio dependency")
if not hasattr(CRS, 'from_dict'):
return unittest.skip("RasterIO 1.0+ required")
import numpy as np
import xarray as xr
from pyresample import parse_area_file
from pyresample.geometry import SwathDefinition
from satpy.resample import get_area_file
lons = np.array([[0, 0.1, 0.2], [0.05, 0.15, 0.25]])
lats = np.array([[0, 0.1, 0.2], [0.05, 0.15, 0.25]])
lons = xr.DataArray(lons)
lats = xr.DataArray(lats)
swath_def = SwathDefinition(lons, lats)
all_areas = parse_area_file(get_area_file())
for area_obj in all_areas:
if hasattr(area_obj, 'freeze'):
try:
area_obj = area_obj.freeze(lonslats=swath_def)
except RuntimeError:
# we didn't provide enough info to freeze, hard to guess
# in a generic test so just skip this area
continue
proj_dict = area_obj.proj_dict
if proj_dict.get('proj') in ('ob_tran', 'nsper') and \
'wktext' not in proj_dict:
# FIXME: rasterio doesn't understand ob_tran unless +wktext
# See: https://github.com/pyproj4/pyproj/issues/357
# pyproj 2.0+ seems to drop wktext from PROJ dict
continue
_ = CRS.from_dict(proj_dict)
[docs]@contextlib.contextmanager
def fake_plugin_etc_path(
tmp_path: Path,
entry_point_names: dict[str, list[str]],
) -> Iterator[Path]:
"""Create a fake satpy plugin entry point.
This mocks the necessary methods to trick Satpy into thinking a plugin
package is installed and has made a satpy plugin available.
"""
etc_path, entry_points = _get_entry_point_list(tmp_path, entry_point_names)
fake_iter_entry_points = _create_fake_iter_entry_points(entry_points)
with mock.patch('satpy._config.pkg_resources.iter_entry_points', fake_iter_entry_points):
yield etc_path
def _get_entry_point_list(
tmp_path: Path,
entry_point_names: dict[str, list[str]]
) -> tuple[Path, dict[str, list[pkg_resources.EntryPoint]]]:
dist_obj = pkg_resources.Distribution.from_filename('satpy_plugin-0.0.0-py3.8.egg')
etc_path = tmp_path / "satpy_plugin" / "etc"
etc_path.mkdir(parents=True, exist_ok=True)
entry_points: dict[str, list[pkg_resources.EntryPoint]] = {}
for entry_point_name, entry_point_values in entry_point_names.items():
entry_points[entry_point_name] = []
for entry_point_value in entry_point_values:
ep = pkg_resources.EntryPoint.parse(entry_point_value)
ep.dist = dist_obj
ep.dist.module_path = tmp_path # type: ignore
entry_points[entry_point_name].append(ep)
return etc_path, entry_points
def _create_fake_iter_entry_points(entry_points: dict[str, list[pkg_resources.EntryPoint]]) -> Callable[[str], list]:
def _fake_iter_entry_points(desired_entry_point_name: str) -> list:
return entry_points.get(desired_entry_point_name, [])
return _fake_iter_entry_points
[docs]@pytest.fixture
def fake_composite_plugin_etc_path(tmp_path: Path) -> Iterator[Path]:
"""Create a fake plugin entry point with a fake compositor YAML configuration file."""
yield from _create_yamlbased_plugin(
tmp_path,
"composites",
"fake_sensor.yaml",
_write_fake_composite_yaml,
)
def _write_fake_composite_yaml(yaml_filename: str) -> None:
with open(yaml_filename, "w") as comps_file:
comps_file.write("""
sensor_name: visir/fake_sensor
composites:
fake_composite:
compositor: !!python/name:satpy.composites.GenericCompositor
prerequisites:
- 3.9
- 10.8
- 12.0
standard_name: fake composite
""")
[docs]@pytest.fixture
def fake_reader_plugin_etc_path(tmp_path: Path) -> Iterator[Path]:
"""Create a fake plugin entry point with a fake reader YAML configuration file."""
yield from _create_yamlbased_plugin(
tmp_path,
"readers",
"fake_reader.yaml",
_write_fake_reader_yaml,
)
def _write_fake_reader_yaml(yaml_filename: str) -> None:
reader_name = os.path.splitext(os.path.basename(yaml_filename))[0]
with open(yaml_filename, "w") as comps_file:
comps_file.write(f"""
reader:
name: {reader_name}
sensors: [fake_sensor]
reader: !!python/name:satpy.readers.yaml_reader.FileYAMLReader
datasets: {{}}
""")
[docs]@pytest.fixture
def fake_writer_plugin_etc_path(tmp_path: Path) -> Iterator[Path]:
"""Create a fake plugin entry point with a fake writer YAML configuration file."""
yield from _create_yamlbased_plugin(
tmp_path,
"writers",
"fake_writer.yaml",
_write_fake_writer_yaml,
)
def _write_fake_writer_yaml(yaml_filename: str) -> None:
writer_name = os.path.splitext(os.path.basename(yaml_filename))[0]
with open(yaml_filename, "w") as comps_file:
comps_file.write(f"""
writer:
name: {writer_name}
writer: !!python/name:satpy.writers.Writer
""")
[docs]@pytest.fixture
def fake_enh_plugin_etc_path(tmp_path: Path) -> Iterator[Path]:
"""Create a fake plugin entry point with a fake enhancement YAML configure files.
This creates a ``fake_sensor.yaml`` and ``generic.yaml`` enhancement configuration.
"""
yield from _create_yamlbased_plugin(
tmp_path,
"enhancements",
"fake_sensor.yaml",
_write_fake_enh_yamls,
)
def _write_fake_enh_yamls(yaml_filename: str) -> None:
with open(yaml_filename, "w") as comps_file:
comps_file.write("""
enhancements:
some_custom_plugin_enh:
name: fake_name
operations:
- name: stretch
method: !!python/name:satpy.enhancements.stretch
kwargs:
stretch: crude
min_stretch: -100.0
max_stretch: 0.0
""")
generic_filename = os.path.join(os.path.dirname(yaml_filename), "generic.yaml")
with open(generic_filename, "w") as comps_file:
comps_file.write("""
enhancements:
default:
operations:
- name: stretch
method: !!python/name:satpy.enhancements.stretch
kwargs:
stretch: crude
min_stretch: -1.0
max_stretch: 1.0
""")
def _create_yamlbased_plugin(
tmp_path: Path,
component_type: str,
yaml_name: str,
yaml_func: Callable[[str], None]
) -> Iterator[Path]:
entry_point_dict = {f"satpy.{component_type}": [f"example_{component_type} = satpy_plugin"]}
with fake_plugin_etc_path(tmp_path, entry_point_dict) as plugin_etc_path:
comps_dir = os.path.join(plugin_etc_path, component_type)
os.makedirs(comps_dir, exist_ok=True)
comps_filename = os.path.join(comps_dir, yaml_name)
yaml_func(comps_filename)
yield plugin_etc_path
[docs]class TestPluginsConfigs:
"""Test that plugins are working."""
[docs] def test_get_plugin_configs(self, fake_composite_plugin_etc_path):
"""Check that the plugin configs are looked for."""
from satpy._config import get_entry_points_config_dirs
with satpy.config.set(config_path=[]):
dirs = get_entry_points_config_dirs('satpy.composites')
assert dirs == [str(fake_composite_plugin_etc_path)]
[docs] def test_load_entry_point_composite(self, fake_composite_plugin_etc_path):
"""Test that composites can be loaded from plugin entry points."""
with satpy.config.set(config_path=[]):
compositors, _ = load_compositor_configs_for_sensors(["fake_sensor"])
assert "fake_sensor" in compositors
comp_dict = DatasetDict(compositors["fake_sensor"])
assert "fake_composite" in comp_dict
comp_obj = comp_dict["fake_composite"]
assert comp_obj.attrs["name"] == "fake_composite"
assert comp_obj.attrs["prerequisites"] == [3.9, 10.8, 12.0]
[docs] @pytest.mark.parametrize("specified_reader", [None, "fake_reader"])
def test_plugin_reader_configs(self, fake_reader_plugin_etc_path, specified_reader):
"""Test that readers can be loaded from plugin entry points."""
from satpy.readers import configs_for_reader
reader_yaml_path = fake_reader_plugin_etc_path / "readers" / "fake_reader.yaml"
self._get_and_check_reader_writer_configs(specified_reader, configs_for_reader, reader_yaml_path)
[docs] def test_plugin_reader_available_readers(self, fake_reader_plugin_etc_path):
"""Test that readers can be loaded from plugin entry points."""
from satpy.readers import available_readers
self._check_available_component(available_readers, "fake_reader")
[docs] @pytest.mark.parametrize("specified_writer", [None, "fake_writer"])
def test_plugin_writer_configs(self, fake_writer_plugin_etc_path, specified_writer):
"""Test that writers can be loaded from plugin entry points."""
from satpy.writers import configs_for_writer
writer_yaml_path = fake_writer_plugin_etc_path / "writers" / "fake_writer.yaml"
self._get_and_check_reader_writer_configs(specified_writer, configs_for_writer, writer_yaml_path)
[docs] def test_plugin_writer_available_writers(self, fake_writer_plugin_etc_path):
"""Test that readers can be loaded from plugin entry points."""
from satpy.writers import available_writers
self._check_available_component(available_writers, "fake_writer")
@staticmethod
def _get_and_check_reader_writer_configs(specified_component, configs_func, exp_yaml):
with satpy.config.set(config_path=[]):
configs = list(configs_func(specified_component))
assert any(str(exp_yaml) in config_list for config_list in configs)
@staticmethod
def _check_available_component(available_func, exp_component):
with satpy.config.set(config_path=[]):
available_components = available_func()
assert exp_component in available_components
[docs] @pytest.mark.parametrize(
("sensor_name", "exp_result"),
[
("fake_sensor", 1.0), # uses the sensor specific entry
("fake_sensor2", 0.5), # uses the generic.yaml default
]
)
def test_plugin_enhancements_generic_sensor(self, fake_enh_plugin_etc_path, sensor_name, exp_result):
"""Test that enhancements from a plugin are available."""
import dask.array as da
import numpy as np
import xarray as xr
from trollimage.xrimage import XRImage
from satpy.writers import Enhancer
data_arr = xr.DataArray(
da.zeros((10, 10), dtype=np.float32),
dims=("y", "x"),
attrs={
"sensor": {sensor_name},
"name": "fake_name",
})
img = XRImage(data_arr)
enh = Enhancer()
enh.add_sensor_enhancements(data_arr.attrs["sensor"])
enh.apply(img, **img.data.attrs)
res_data = img.data.values
np.testing.assert_allclose(res_data, exp_result)
[docs]class TestConfigObject:
"""Test basic functionality of the central config object."""
[docs] def test_custom_config_file(self):
"""Test adding a custom configuration file using SATPY_CONFIG."""
import tempfile
from importlib import reload
import yaml
import satpy
my_config_dict = {
'cache_dir': "/path/to/cache",
}
try:
with tempfile.NamedTemporaryFile(mode='w+t', suffix='.yaml', delete=False) as tfile:
yaml.dump(my_config_dict, tfile)
tfile.close()
with mock.patch.dict('os.environ', {'SATPY_CONFIG': tfile.name}):
reload(satpy._config)
reload(satpy)
assert satpy.config.get('cache_dir') == '/path/to/cache'
finally:
os.remove(tfile.name)
[docs] def test_deprecated_env_vars(self):
"""Test that deprecated variables are mapped to new config."""
from importlib import reload
import satpy
old_vars = {
'PPP_CONFIG_DIR': '/my/ppp/config/dir',
'SATPY_ANCPATH': '/my/ancpath',
}
with mock.patch.dict('os.environ', old_vars):
reload(satpy._config)
reload(satpy)
assert satpy.config.get('data_dir') == '/my/ancpath'
assert satpy.config.get('config_path') == ['/my/ppp/config/dir']
[docs] def test_config_path_multiple(self):
"""Test that multiple config paths are accepted."""
from importlib import reload
import satpy
exp_paths, env_paths = _os_specific_multipaths()
old_vars = {
'SATPY_CONFIG_PATH': env_paths,
}
with mock.patch.dict('os.environ', old_vars):
reload(satpy._config)
reload(satpy)
assert satpy.config.get('config_path') == exp_paths
[docs] def test_config_path_multiple_load(self):
"""Test that config paths from subprocesses load properly.
Satpy modifies the config path environment variable when it is imported.
If Satpy is imported again from a subprocess then it should be able to parse this
modified variable.
"""
from importlib import reload
import satpy
exp_paths, env_paths = _os_specific_multipaths()
old_vars = {
'SATPY_CONFIG_PATH': env_paths,
}
with mock.patch.dict('os.environ', old_vars):
# these reloads will update env variable "SATPY_CONFIG_PATH"
reload(satpy._config)
reload(satpy)
# load the updated env variable and parse it again.
reload(satpy._config)
reload(satpy)
assert satpy.config.get('config_path') == exp_paths
[docs] def test_bad_str_config_path(self):
"""Test that a str config path isn't allowed."""
from importlib import reload
import satpy
old_vars = {
'SATPY_CONFIG_PATH': '/my/configs1',
}
# single path from env var still works
with mock.patch.dict('os.environ', old_vars):
reload(satpy._config)
reload(satpy)
assert satpy.config.get('config_path') == ['/my/configs1']
# strings are not allowed, lists are
with satpy.config.set(config_path='/single/string/paths/are/bad'):
pytest.raises(ValueError, satpy._config.get_config_path_safe)
def _os_specific_multipaths():
exp_paths = ['/my/configs1', '/my/configs2', '/my/configs3']
if sys.platform.startswith("win"):
exp_paths = ["C:" + p for p in exp_paths]
path_str = os.pathsep.join(exp_paths)
return exp_paths, path_str