Source code for satpy.tests.test_resample

#!/usr/bin/python
# Copyright (c) 2016 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/>.
"""Unittests for resamplers."""

import os
import shutil
import tempfile
import unittest
from unittest import mock

try:
    from pyresample.ewa import LegacyDaskEWAResampler
except ImportError:
    LegacyDaskEWAResampler = None

from pyproj import CRS


[docs]def get_test_data(input_shape=(100, 50), output_shape=(200, 100), output_proj=None, input_dims=('y', 'x')): """Get common data objects used in testing. Returns: tuple: * input_data_on_area: DataArray with dimensions as if it is a gridded dataset. * input_area_def: AreaDefinition of the above DataArray * input_data_on_swath: DataArray with dimensions as if it is a swath. * input_swath: SwathDefinition of the above DataArray * target_area_def: AreaDefinition to be used as a target for resampling """ import dask.array as da from pyresample.geometry import AreaDefinition, SwathDefinition from pyresample.utils import proj4_str_to_dict from xarray import DataArray ds1 = DataArray(da.zeros(input_shape, chunks=85), dims=input_dims, attrs={'name': 'test_data_name', 'test': 'test'}) if input_dims and 'y' in input_dims: ds1 = ds1.assign_coords(y=da.arange(input_shape[-2], chunks=85)) if input_dims and 'x' in input_dims: ds1 = ds1.assign_coords(x=da.arange(input_shape[-1], chunks=85)) if input_dims and 'bands' in input_dims: ds1 = ds1.assign_coords(bands=list('RGBA'[:ds1.sizes['bands']])) input_proj_str = ('+proj=geos +lon_0=-95.0 +h=35786023.0 +a=6378137.0 ' '+b=6356752.31414 +sweep=x +units=m +no_defs') source = AreaDefinition( 'test_target', 'test_target', 'test_target', proj4_str_to_dict(input_proj_str), input_shape[1], # width input_shape[0], # height (-1000., -1500., 1000., 1500.)) ds1.attrs['area'] = source crs = CRS.from_string(input_proj_str) ds1 = ds1.assign_coords(crs=crs) ds2 = ds1.copy() input_area_shape = tuple(ds1.sizes[dim] for dim in ds1.dims if dim in ['y', 'x']) geo_dims = ('y', 'x') if input_dims else None lons = da.random.random(input_area_shape, chunks=50) lats = da.random.random(input_area_shape, chunks=50) swath_def = SwathDefinition( DataArray(lons, dims=geo_dims), DataArray(lats, dims=geo_dims)) ds2.attrs['area'] = swath_def crs = CRS.from_string('+proj=latlong +datum=WGS84 +ellps=WGS84') ds2 = ds2.assign_coords(crs=crs) # set up target definition output_proj_str = ('+proj=lcc +datum=WGS84 +ellps=WGS84 ' '+lon_0=-95. +lat_0=25 +lat_1=25 +units=m +no_defs') output_proj_str = output_proj or output_proj_str target = AreaDefinition( 'test_target', 'test_target', 'test_target', proj4_str_to_dict(output_proj_str), output_shape[1], # width output_shape[0], # height (-1000., -1500., 1000., 1500.), ) return ds1, source, ds2, swath_def, target
[docs]class TestHLResample(unittest.TestCase): """Test the higher level resampling functions."""
[docs] def test_type_preserve(self): """Check that the type of resampled datasets is preserved.""" import dask.array as da import numpy as np import xarray as xr from pyresample.geometry import SwathDefinition from satpy.resample import resample_dataset source_area = SwathDefinition(xr.DataArray(da.arange(4, chunks=5).reshape((2, 2)), dims=['y', 'x']), xr.DataArray(da.arange(4, chunks=5).reshape((2, 2)), dims=['y', 'x'])) dest_area = SwathDefinition(xr.DataArray(da.arange(4, chunks=5).reshape((2, 2)) + .0001, dims=['y', 'x']), xr.DataArray(da.arange(4, chunks=5).reshape((2, 2)) + .0001, dims=['y', 'x'])) expected_gap = np.array([[1, 2], [3, 255]]) data = xr.DataArray(da.from_array(expected_gap, chunks=5), dims=['y', 'x']) data.attrs['_FillValue'] = 255 data.attrs['area'] = source_area res = resample_dataset(data, dest_area) self.assertEqual(res.dtype, data.dtype) self.assertTrue(np.all(res.values == expected_gap)) expected_filled = np.array([[1, 2], [3, 3]]) res = resample_dataset(data, dest_area, radius_of_influence=1000000) self.assertEqual(res.dtype, data.dtype) self.assertTrue(np.all(res.values == expected_filled))
[docs]class TestKDTreeResampler(unittest.TestCase): """Test the kd-tree resampler."""
[docs] @mock.patch('satpy.resample.KDTreeResampler._check_numpy_cache') @mock.patch('satpy.resample.xr.Dataset') @mock.patch('satpy.resample.zarr.open') @mock.patch('satpy.resample.KDTreeResampler._create_cache_filename') @mock.patch('pyresample.kd_tree.XArrayResamplerNN') def test_kd_resampling(self, xr_resampler, create_filename, zarr_open, xr_dset, cnc): """Test the kd resampler.""" import dask.array as da from satpy.resample import KDTreeResampler data, source_area, swath_data, source_swath, target_area = get_test_data() mock_dset = mock.MagicMock() xr_dset.return_value = mock_dset resampler = KDTreeResampler(source_swath, target_area) resampler.precompute( mask=da.arange(5, chunks=5).astype(bool), cache_dir='.') xr_resampler.assert_called_once() resampler.resampler.get_neighbour_info.assert_called() # swath definitions should not be cached self.assertFalse(len(mock_dset.to_zarr.mock_calls), 0) resampler.resampler.reset_mock() cnc.assert_called_once() resampler = KDTreeResampler(source_area, target_area) resampler.precompute() resampler.resampler.get_neighbour_info.assert_called_with(mask=None) try: the_dir = tempfile.mkdtemp() resampler = KDTreeResampler(source_area, target_area) create_filename.return_value = os.path.join(the_dir, 'test_cache.zarr') zarr_open.side_effect = ValueError() resampler.precompute(cache_dir=the_dir) # assert data was saved to the on-disk cache self.assertEqual(len(mock_dset.to_zarr.mock_calls), 1) # assert that zarr_open was called to try to zarr_open something from disk self.assertEqual(len(zarr_open.mock_calls), 1) # we should have cached things in-memory self.assertEqual(len(resampler._index_caches), 1) nbcalls = len(resampler.resampler.get_neighbour_info.mock_calls) # test reusing the resampler zarr_open.side_effect = None # The kdtree shouldn't be available after saving cache to disk assert resampler.resampler.delayed_kdtree is None class FakeZarr(dict): def close(self): pass def astype(self, dtype): pass zarr_open.return_value = FakeZarr(valid_input_index=1, valid_output_index=2, index_array=3, distance_array=4) resampler.precompute(cache_dir=the_dir) # we already have things cached in-memory, no need to save again self.assertEqual(len(mock_dset.to_zarr.mock_calls), 1) # we already have things cached in-memory, don't need to load self.assertEqual(len(zarr_open.mock_calls), 1) # we should have cached things in-memory self.assertEqual(len(resampler._index_caches), 1) self.assertEqual(len(resampler.resampler.get_neighbour_info.mock_calls), nbcalls) # test loading saved resampler resampler = KDTreeResampler(source_area, target_area) resampler.precompute(cache_dir=the_dir) self.assertEqual(len(zarr_open.mock_calls), 4) self.assertEqual(len(resampler.resampler.get_neighbour_info.mock_calls), nbcalls) # we should have cached things in-memory now self.assertEqual(len(resampler._index_caches), 1) finally: shutil.rmtree(the_dir) fill_value = 8 resampler.compute(data, fill_value=fill_value) resampler.resampler.get_sample_from_neighbour_info.assert_called_with(data, fill_value)
[docs] @mock.patch('satpy.resample.np.load') @mock.patch('satpy.resample.xr.Dataset') def test_check_numpy_cache(self, xr_Dataset, np_load): """Test that cache stored in .npz is converted to zarr.""" from satpy.resample import KDTreeResampler data, source_area, swath_data, source_swath, target_area = get_test_data() resampler = KDTreeResampler(source_area, target_area) zarr_out = mock.MagicMock() xr_Dataset.return_value = zarr_out try: the_dir = tempfile.mkdtemp() kwargs = {} np_path = resampler._create_cache_filename(the_dir, prefix='resample_lut-', fmt='.npz', mask=None, **kwargs) zarr_path = resampler._create_cache_filename(the_dir, prefix='nn_lut-', fmt='.zarr', mask=None, **kwargs) resampler._check_numpy_cache(the_dir) np_load.assert_not_called() zarr_out.to_zarr.assert_not_called() with open(np_path, 'w') as fid: fid.write("42") resampler._check_numpy_cache(the_dir) np_load.assert_called_once_with(np_path, 'r') zarr_out.to_zarr.assert_called_once_with(zarr_path) finally: shutil.rmtree(the_dir)
[docs]@unittest.skipIf(LegacyDaskEWAResampler is not None, "Deprecated EWA resampler is now in pyresample. " "No need to test in Satpy.") class TestEWAResampler(unittest.TestCase): """Test EWA resampler class."""
[docs] @mock.patch('satpy.resample.fornav') @mock.patch('satpy.resample.ll2cr') @mock.patch('satpy.resample.SwathDefinition.get_lonlats') def test_2d_ewa(self, get_lonlats, ll2cr, fornav): """Test EWA with a 2D dataset.""" import numpy as np import xarray as xr from satpy.resample import resample_dataset ll2cr.return_value = (100, np.zeros((10, 10), dtype=np.float32), np.zeros((10, 10), dtype=np.float32)) fornav.return_value = (100 * 200, np.zeros((200, 100), dtype=np.float32)) _, _, swath_data, source_swath, target_area = get_test_data() get_lonlats.return_value = (source_swath.lons, source_swath.lats) swath_data.data = swath_data.data.astype(np.float32) num_chunks = len(source_swath.lons.chunks[0]) * len(source_swath.lons.chunks[1]) new_data = resample_dataset(swath_data, target_area, resampler='ewa') self.assertTupleEqual(new_data.shape, (200, 100)) self.assertEqual(new_data.dtype, np.float32) self.assertEqual(new_data.attrs['test'], 'test') self.assertIs(new_data.attrs['area'], target_area) # make sure we can actually compute everything new_data.compute() lonlat_calls = get_lonlats.call_count ll2cr_calls = ll2cr.call_count # resample a different dataset and make sure cache is used data = xr.DataArray( swath_data.data, dims=('y', 'x'), attrs={'area': source_swath, 'test': 'test2', 'name': 'test2'}) new_data = resample_dataset(data, target_area, resampler='ewa') new_data.compute() # ll2cr will be called once more because of the computation self.assertEqual(ll2cr.call_count, ll2cr_calls + num_chunks) # but we should already have taken the lonlats from the SwathDefinition self.assertEqual(get_lonlats.call_count, lonlat_calls) self.assertIn('y', new_data.coords) self.assertIn('x', new_data.coords) self.assertIn('crs', new_data.coords) self.assertIsInstance(new_data.coords['crs'].item(), CRS) self.assertIn('lambert', new_data.coords['crs'].item().coordinate_operation.method_name.lower()) self.assertEqual(new_data.coords['y'].attrs['units'], 'meter') self.assertEqual(new_data.coords['x'].attrs['units'], 'meter') self.assertEqual(target_area.crs, new_data.coords['crs'].item())
[docs] @mock.patch('satpy.resample.fornav') @mock.patch('satpy.resample.ll2cr') @mock.patch('satpy.resample.SwathDefinition.get_lonlats') def test_3d_ewa(self, get_lonlats, ll2cr, fornav): """Test EWA with a 3D dataset.""" import numpy as np import xarray as xr from satpy.resample import resample_dataset _, _, swath_data, source_swath, target_area = get_test_data( input_shape=(3, 200, 100), input_dims=('bands', 'y', 'x')) swath_data.data = swath_data.data.astype(np.float32) ll2cr.return_value = (100, np.zeros((10, 10), dtype=np.float32), np.zeros((10, 10), dtype=np.float32)) fornav.return_value = ([100 * 200] * 3, [np.zeros((200, 100), dtype=np.float32)] * 3) get_lonlats.return_value = (source_swath.lons, source_swath.lats) num_chunks = len(source_swath.lons.chunks[0]) * len(source_swath.lons.chunks[1]) new_data = resample_dataset(swath_data, target_area, resampler='ewa') self.assertTupleEqual(new_data.shape, (3, 200, 100)) self.assertEqual(new_data.dtype, np.float32) self.assertEqual(new_data.attrs['test'], 'test') self.assertIs(new_data.attrs['area'], target_area) # make sure we can actually compute everything new_data.compute() lonlat_calls = get_lonlats.call_count ll2cr_calls = ll2cr.call_count # resample a different dataset and make sure cache is used swath_data = xr.DataArray( swath_data.data, dims=('bands', 'y', 'x'), coords={'bands': ['R', 'G', 'B']}, attrs={'area': source_swath, 'test': 'test'}) new_data = resample_dataset(swath_data, target_area, resampler='ewa') new_data.compute() # ll2cr will be called once more because of the computation self.assertEqual(ll2cr.call_count, ll2cr_calls + num_chunks) # but we should already have taken the lonlats from the SwathDefinition self.assertEqual(get_lonlats.call_count, lonlat_calls) self.assertIn('y', new_data.coords) self.assertIn('x', new_data.coords) self.assertIn('bands', new_data.coords) self.assertIn('crs', new_data.coords) self.assertIsInstance(new_data.coords['crs'].item(), CRS) self.assertIn('lambert', new_data.coords['crs'].item().coordinate_operation.method_name.lower()) self.assertEqual(new_data.coords['y'].attrs['units'], 'meter') self.assertEqual(new_data.coords['x'].attrs['units'], 'meter') np.testing.assert_equal(new_data.coords['bands'].values, ['R', 'G', 'B']) self.assertEqual(target_area.crs, new_data.coords['crs'].item())
[docs]class TestNativeResampler(unittest.TestCase): """Tests for the 'native' resampling method."""
[docs] def test_expand_reduce(self): """Test class method 'expand_reduce' basics.""" import dask.array as da import numpy as np from satpy.resample import NativeResampler d_arr = da.zeros((6, 20), chunks=4) new_data = NativeResampler._expand_reduce(d_arr, {0: 2., 1: 2.}) self.assertEqual(new_data.shape, (12, 40)) new_data = NativeResampler._expand_reduce(d_arr, {0: .5, 1: .5}) self.assertEqual(new_data.shape, (3, 10)) self.assertRaises(ValueError, NativeResampler._expand_reduce, d_arr, {0: 1. / 3, 1: 1.}) new_data = NativeResampler._expand_reduce(d_arr, {0: 1., 1: 1.}) self.assertEqual(new_data.shape, (6, 20)) self.assertIs(new_data, d_arr) self.assertRaises(ValueError, NativeResampler._expand_reduce, d_arr, {0: 0.333323423, 1: 1.}) self.assertRaises(ValueError, NativeResampler._expand_reduce, d_arr, {0: 1.333323423, 1: 1.}) n_arr = np.zeros((6, 20)) new_data = NativeResampler._expand_reduce(n_arr, {0: 2., 1: 1.0}) self.assertTrue(np.all(new_data.compute()[::2, :] == n_arr))
[docs] def test_expand_dims(self): """Test expanding native resampling with 2D data.""" import numpy as np from satpy.resample import NativeResampler ds1, source_area, _, _, target_area = get_test_data() # source geo def doesn't actually matter resampler = NativeResampler(source_area, target_area) new_data = resampler.resample(ds1) self.assertEqual(new_data.shape, (200, 100)) new_data2 = resampler.resample(ds1.compute()) self.assertTrue(np.all(new_data == new_data2)) self.assertIn('y', new_data.coords) self.assertIn('x', new_data.coords) self.assertIn('crs', new_data.coords) self.assertIsInstance(new_data.coords['crs'].item(), CRS) self.assertIn('lambert', new_data.coords['crs'].item().coordinate_operation.method_name.lower()) self.assertEqual(new_data.coords['y'].attrs['units'], 'meter') self.assertEqual(new_data.coords['x'].attrs['units'], 'meter') self.assertEqual(target_area.crs, new_data.coords['crs'].item())
[docs] def test_expand_dims_3d(self): """Test expanding native resampling with 3D data.""" import numpy as np from satpy.resample import NativeResampler ds1, source_area, _, _, target_area = get_test_data( input_shape=(3, 100, 50), input_dims=('bands', 'y', 'x')) # source geo def doesn't actually matter resampler = NativeResampler(source_area, target_area) new_data = resampler.resample(ds1) self.assertEqual(new_data.shape, (3, 200, 100)) new_data2 = resampler.resample(ds1.compute()) self.assertTrue(np.all(new_data == new_data2)) self.assertIn('y', new_data.coords) self.assertIn('x', new_data.coords) self.assertIn('bands', new_data.coords) np.testing.assert_equal(new_data.coords['bands'].values, ['R', 'G', 'B']) self.assertIn('crs', new_data.coords) self.assertIsInstance(new_data.coords['crs'].item(), CRS) self.assertIn('lambert', new_data.coords['crs'].item().coordinate_operation.method_name.lower()) self.assertEqual(new_data.coords['y'].attrs['units'], 'meter') self.assertEqual(new_data.coords['x'].attrs['units'], 'meter') self.assertEqual(target_area.crs, new_data.coords['crs'].item())
[docs] def test_expand_without_dims(self): """Test expanding native resampling with no dimensions specified.""" import numpy as np from satpy.resample import NativeResampler ds1, source_area, _, _, target_area = get_test_data(input_dims=None) # source geo def doesn't actually matter resampler = NativeResampler(source_area, target_area) new_data = resampler.resample(ds1) self.assertEqual(new_data.shape, (200, 100)) new_data2 = resampler.resample(ds1.compute()) self.assertTrue(np.all(new_data == new_data2)) self.assertIn('crs', new_data.coords) self.assertIsInstance(new_data.coords['crs'].item(), CRS) self.assertIn('lambert', new_data.coords['crs'].item().coordinate_operation.method_name.lower()) self.assertEqual(target_area.crs, new_data.coords['crs'].item())
[docs] def test_expand_without_dims_4D(self): """Test expanding native resampling with 4D data with no dimensions specified.""" from satpy.resample import NativeResampler ds1, source_area, _, _, target_area = get_test_data( input_shape=(2, 3, 100, 50), input_dims=None) # source geo def doesn't actually matter resampler = NativeResampler(source_area, target_area) self.assertRaises(ValueError, resampler.resample, ds1)
[docs]class TestBilinearResampler(unittest.TestCase): """Test the bilinear resampler."""
[docs] @mock.patch('satpy.resample._move_existing_caches') @mock.patch('satpy.resample.BilinearResampler._create_cache_filename') @mock.patch('pyresample.bilinear.XArrayBilinearResampler') def test_bil_resampling(self, xr_resampler, create_filename, move_existing_caches): """Test the bilinear resampler.""" import dask.array as da import xarray as xr from satpy.resample import BilinearResampler data, source_area, swath_data, source_swath, target_area = get_test_data() # Test that bilinear resampling info calculation is called resampler = BilinearResampler(source_swath, target_area) resampler.precompute( mask=da.arange(5, chunks=5).astype(bool)) resampler.resampler.load_resampling_info.assert_not_called() resampler.resampler.get_bil_info.assert_called_once() resampler.resampler.reset_mock() # Test that get_sample_from_bil_info is called properly fill_value = 8 resampler.resampler.get_sample_from_bil_info.return_value = \ xr.DataArray(da.zeros(target_area.shape), dims=('y', 'x')) new_data = resampler.compute(data, fill_value=fill_value) resampler.resampler.get_sample_from_bil_info.assert_called_with( data, fill_value=fill_value, output_shape=target_area.shape) self.assertIn('y', new_data.coords) self.assertIn('x', new_data.coords) self.assertIn('crs', new_data.coords) self.assertIsInstance(new_data.coords['crs'].item(), CRS) self.assertIn('lambert', new_data.coords['crs'].item().coordinate_operation.method_name.lower()) self.assertEqual(new_data.coords['y'].attrs['units'], 'meter') self.assertEqual(new_data.coords['x'].attrs['units'], 'meter') self.assertEqual(target_area.crs, new_data.coords['crs'].item()) # Test that the resampling info is tried to read from the disk resampler = BilinearResampler(source_swath, target_area) resampler.precompute(cache_dir='.') resampler.resampler.load_resampling_info.assert_called() # Test caching the resampling info try: the_dir = tempfile.mkdtemp() resampler = BilinearResampler(source_area, target_area) create_filename.return_value = os.path.join(the_dir, 'test_cache.zarr') xr_resampler.return_value.load_resampling_info.side_effect = IOError resampler.precompute(cache_dir=the_dir) resampler.resampler.save_resampling_info.assert_called() # assert data was saved to the on-disk cache resampler.resampler.save_resampling_info.assert_called_once() nbcalls = resampler.resampler.get_bil_info.call_count resampler.resampler.load_resampling_info.side_effect = None resampler.precompute(cache_dir=the_dir) # we already have things cached in-memory, no need to save again resampler.resampler.save_resampling_info.assert_called_once() # we already have things cached in-memory, don't need to load self.assertEqual(resampler.resampler.get_bil_info.call_count, nbcalls) # test loading saved resampler resampler = BilinearResampler(source_area, target_area) resampler.precompute(cache_dir=the_dir) self.assertEqual(resampler.resampler.load_resampling_info.call_count, 3) self.assertEqual(resampler.resampler.get_bil_info.call_count, nbcalls) resampler = BilinearResampler(source_area, target_area) resampler.precompute(cache_dir=the_dir) resampler.save_bil_info(cache_dir=the_dir) zarr_file = os.path.join(the_dir, 'test_cache.zarr') # Save again faking the cache file already exists with mock.patch('os.path.exists') as exists: exists.return_value = True resampler.save_bil_info(cache_dir=the_dir) move_existing_caches.assert_called_once_with(the_dir, zarr_file) finally: shutil.rmtree(the_dir)
[docs] def test_move_existing_caches(self): """Test that existing caches are moved to a subdirectory.""" try: the_dir = tempfile.mkdtemp() # Test that existing cache file is moved away zarr_file = os.path.join(the_dir, 'test.zarr') with open(zarr_file, 'w') as fid: fid.write('42') from satpy.resample import _move_existing_caches _move_existing_caches(the_dir, zarr_file) self.assertFalse(os.path.exists(zarr_file)) self.assertTrue(os.path.exists( os.path.join(the_dir, 'moved_by_satpy', 'test.zarr'))) # Run again to see that the existing dir doesn't matter with open(zarr_file, 'w') as fid: fid.write('42') _move_existing_caches(the_dir, zarr_file) finally: shutil.rmtree(the_dir)
[docs]class TestCoordinateHelpers(unittest.TestCase): """Test various utility functions for working with coordinates."""
[docs] def test_area_def_coordinates(self): """Test coordinates being added with an AreaDefinition.""" import dask.array as da import numpy as np import xarray as xr from pyresample.geometry import AreaDefinition from satpy.resample import add_crs_xy_coords area_def = AreaDefinition( 'test', 'test', 'test', {'proj': 'lcc', 'lat_1': 25, 'lat_0': 25}, 100, 200, [-100, -100, 100, 100] ) data_arr = xr.DataArray( da.zeros((200, 100), chunks=50), attrs={'area': area_def}, dims=('y', 'x'), ) new_data_arr = add_crs_xy_coords(data_arr, area_def) self.assertIn('y', new_data_arr.coords) self.assertIn('x', new_data_arr.coords) self.assertIn('units', new_data_arr.coords['y'].attrs) self.assertEqual( new_data_arr.coords['y'].attrs['units'], 'meter') self.assertIn('units', new_data_arr.coords['x'].attrs) self.assertEqual( new_data_arr.coords['x'].attrs['units'], 'meter') self.assertIn('crs', new_data_arr.coords) self.assertIsInstance(new_data_arr.coords['crs'].item(), CRS) self.assertEqual(area_def.crs, new_data_arr.coords['crs'].item()) # already has coords data_arr = xr.DataArray( da.zeros((200, 100), chunks=50), attrs={'area': area_def}, dims=('y', 'x'), coords={'y': np.arange(2, 202), 'x': np.arange(100)} ) new_data_arr = add_crs_xy_coords(data_arr, area_def) self.assertIn('y', new_data_arr.coords) self.assertNotIn('units', new_data_arr.coords['y'].attrs) self.assertIn('x', new_data_arr.coords) self.assertNotIn('units', new_data_arr.coords['x'].attrs) np.testing.assert_equal(new_data_arr.coords['y'], np.arange(2, 202)) self.assertIn('crs', new_data_arr.coords) self.assertIsInstance(new_data_arr.coords['crs'].item(), CRS) self.assertEqual(area_def.crs, new_data_arr.coords['crs'].item()) # lat/lon area area_def = AreaDefinition( 'test', 'test', 'test', {'proj': 'latlong'}, 100, 200, [-100, -100, 100, 100] ) data_arr = xr.DataArray( da.zeros((200, 100), chunks=50), attrs={'area': area_def}, dims=('y', 'x'), ) new_data_arr = add_crs_xy_coords(data_arr, area_def) self.assertIn('y', new_data_arr.coords) self.assertIn('x', new_data_arr.coords) self.assertIn('units', new_data_arr.coords['y'].attrs) self.assertEqual( new_data_arr.coords['y'].attrs['units'], 'degrees_north') self.assertIn('units', new_data_arr.coords['x'].attrs) self.assertEqual( new_data_arr.coords['x'].attrs['units'], 'degrees_east') self.assertIn('crs', new_data_arr.coords) self.assertIsInstance(new_data_arr.coords['crs'].item(), CRS) self.assertEqual(area_def.crs, new_data_arr.coords['crs'].item())
[docs] def test_swath_def_coordinates(self): """Test coordinates being added with an SwathDefinition.""" import dask.array as da import xarray as xr from pyresample.geometry import SwathDefinition from satpy.resample import add_crs_xy_coords lons_data = da.random.random((200, 100), chunks=50) lats_data = da.random.random((200, 100), chunks=50) lons = xr.DataArray(lons_data, attrs={'units': 'degrees_east'}, dims=('y', 'x')) lats = xr.DataArray(lats_data, attrs={'units': 'degrees_north'}, dims=('y', 'x')) area_def = SwathDefinition(lons, lats) data_arr = xr.DataArray( da.zeros((200, 100), chunks=50), attrs={'area': area_def}, dims=('y', 'x'), ) new_data_arr = add_crs_xy_coords(data_arr, area_def) # See https://github.com/pydata/xarray/issues/3068 # self.assertIn('longitude', new_data_arr.coords) # self.assertIn('units', new_data_arr.coords['longitude'].attrs) # self.assertEqual( # new_data_arr.coords['longitude'].attrs['units'], 'degrees_east') # self.assertIsInstance(new_data_arr.coords['longitude'].data, da.Array) # self.assertIn('latitude', new_data_arr.coords) # self.assertIn('units', new_data_arr.coords['latitude'].attrs) # self.assertEqual( # new_data_arr.coords['latitude'].attrs['units'], 'degrees_north') # self.assertIsInstance(new_data_arr.coords['latitude'].data, da.Array) self.assertIn('crs', new_data_arr.coords) crs = new_data_arr.coords['crs'].item() self.assertIsInstance(crs, CRS) assert crs.is_geographic self.assertIsInstance(new_data_arr.coords['crs'].item(), CRS)
[docs]class TestBucketAvg(unittest.TestCase): """Test the bucket resampler."""
[docs] def setUp(self): """Create fake area definitions and resampler to be tested.""" from satpy.resample import BucketAvg get_lonlats = mock.MagicMock() get_lonlats.return_value = (1, 2) get_proj_vectors = mock.MagicMock() get_proj_vectors.return_value = ([1, 2, 3, 4, 5], [1, 2, 3, 4, 5]) self.source_geo_def = mock.MagicMock(get_lonlats=get_lonlats) self.target_geo_def = mock.MagicMock(get_lonlats=get_lonlats, crs=None, get_proj_vectors=get_proj_vectors) self.bucket = BucketAvg(self.source_geo_def, self.target_geo_def)
[docs] def test_init(self): """Test bucket resampler initialization.""" self.assertIsNone(self.bucket.resampler) self.assertTrue(self.bucket.source_geo_def == self.source_geo_def) self.assertTrue(self.bucket.target_geo_def == self.target_geo_def)
[docs] @mock.patch('pyresample.bucket.BucketResampler') def test_precompute(self, bucket): """Test bucket resampler precomputation.""" bucket.return_value = True self.bucket.precompute() self.assertTrue(self.bucket.resampler) bucket.assert_called_once_with(self.target_geo_def, 1, 2)
def _compute_mocked_bucket_avg(self, data, return_data=None, **kwargs): """Compute the mocked bucket average.""" self.bucket.resampler = mock.MagicMock() if return_data is not None: self.bucket.resampler.get_average.return_value = return_data else: self.bucket.resampler.get_average.return_value = data res = self.bucket.compute(data, **kwargs) return res
[docs] def test_compute(self): """Test bucket resampler computation.""" import dask.array as da # 1D data data = da.ones((5,)) res = self._compute_mocked_bucket_avg(data, fill_value=2) self.assertEqual(res.shape, (1, 5)) # 2D data data = da.ones((5, 5)) res = self._compute_mocked_bucket_avg(data, fill_value=2) self.assertEqual(res.shape, (1, 5, 5)) # 3D data data = da.ones((3, 5, 5)) self.bucket.resampler.get_average.return_value = data[0, :, :] res = self._compute_mocked_bucket_avg(data, return_data=data[0, :, :], fill_value=2) self.assertEqual(res.shape, (3, 5, 5))
[docs] @mock.patch('satpy.resample.PR_USE_SKIPNA', True) def test_compute_and_use_skipna_handling(self): """Test bucket resampler computation and use skipna handling.""" import dask.array as da data = da.ones((5,)) self._compute_mocked_bucket_avg(data, fill_value=2, mask_all_nan=True) self.bucket.resampler.get_average.assert_called_once_with( data, fill_value=2, skipna=True) self._compute_mocked_bucket_avg(data, fill_value=2, skipna=False) self.bucket.resampler.get_average.assert_called_once_with( data, fill_value=2, skipna=False) self._compute_mocked_bucket_avg(data, fill_value=2) self.bucket.resampler.get_average.assert_called_once_with( data, fill_value=2, skipna=True)
[docs] @mock.patch('satpy.resample.PR_USE_SKIPNA', False) def test_compute_and_not_use_skipna_handling(self): """Test bucket resampler computation and not use skipna handling.""" import dask.array as da data = da.ones((5,)) self._compute_mocked_bucket_avg(data, fill_value=2, mask_all_nan=True) self.bucket.resampler.get_average.assert_called_once_with( data, fill_value=2, mask_all_nan=True) self._compute_mocked_bucket_avg(data, fill_value=2, mask_all_nan=False) self.bucket.resampler.get_average.assert_called_once_with( data, fill_value=2, mask_all_nan=False) self._compute_mocked_bucket_avg(data, fill_value=2) self.bucket.resampler.get_average.assert_called_once_with( data, fill_value=2, mask_all_nan=False) self._compute_mocked_bucket_avg(data, fill_value=2, skipna=True) self.bucket.resampler.get_average.assert_called_once_with( data, fill_value=2, mask_all_nan=False)
[docs] @mock.patch('pyresample.bucket.BucketResampler') def test_resample(self, pyresample_bucket): """Test bucket resamplers resample method.""" import dask.array as da import xarray as xr self.bucket.resampler = mock.MagicMock() self.bucket.precompute = mock.MagicMock() self.bucket.compute = mock.MagicMock() # 1D input data data = xr.DataArray(da.ones((5,)), dims=('foo'), attrs={'bar': 'baz'}) self.bucket.compute.return_value = da.ones((5, 5)) res = self.bucket.resample(data) self.bucket.precompute.assert_called_once() self.bucket.compute.assert_called_once() self.assertEqual(res.shape, (5, 5)) self.assertEqual(res.dims, ('y', 'x')) self.assertTrue('bar' in res.attrs) self.assertEqual(res.attrs['bar'], 'baz') # 2D input data data = xr.DataArray(da.ones((5, 5)), dims=('foo', 'bar')) self.bucket.compute.return_value = da.ones((5, 5)) res = self.bucket.resample(data) self.assertEqual(res.shape, (5, 5)) self.assertEqual(res.dims, ('y', 'x')) # 3D input data with 'bands' dim data = xr.DataArray(da.ones((1, 5, 5)), dims=('bands', 'foo', 'bar'), coords={'bands': ['L']}) self.bucket.compute.return_value = da.ones((1, 5, 5)) res = self.bucket.resample(data) self.assertEqual(res.shape, (1, 5, 5)) self.assertEqual(res.dims, ('bands', 'y', 'x')) self.assertEqual(res.coords['bands'], ['L']) # 3D input data with misc dim names data = xr.DataArray(da.ones((3, 5, 5)), dims=('foo', 'bar', 'baz')) self.bucket.compute.return_value = da.ones((3, 5, 5)) res = self.bucket.resample(data) self.assertEqual(res.shape, (3, 5, 5)) self.assertEqual(res.dims, ('foo', 'bar', 'baz'))
[docs]class TestBucketSum(unittest.TestCase): """Test the sum bucket resampler."""
[docs] def setUp(self): """Create fake area definitions and resampler to be tested.""" from satpy.resample import BucketSum get_lonlats = mock.MagicMock() get_lonlats.return_value = (1, 2) self.source_geo_def = mock.MagicMock(get_lonlats=get_lonlats) self.target_geo_def = mock.MagicMock(get_lonlats=get_lonlats) self.bucket = BucketSum(self.source_geo_def, self.target_geo_def)
def _compute_mocked_bucket_sum(self, data, return_data=None, **kwargs): """Compute the mocked bucket sum.""" self.bucket.resampler = mock.MagicMock() if return_data is not None: self.bucket.resampler.get_sum.return_value = return_data else: self.bucket.resampler.get_sum.return_value = data res = self.bucket.compute(data, **kwargs) return res
[docs] def test_compute(self): """Test sum bucket resampler computation.""" import dask.array as da # 1D data data = da.ones((5,)) res = self._compute_mocked_bucket_sum(data) self.assertEqual(res.shape, (1, 5)) # 2D data data = da.ones((5, 5)) res = self._compute_mocked_bucket_sum(data) self.assertEqual(res.shape, (1, 5, 5)) # 3D data data = da.ones((3, 5, 5)) res = self._compute_mocked_bucket_sum(data, return_data=data[0, :, :]) self.assertEqual(res.shape, (3, 5, 5))
[docs] @mock.patch('satpy.resample.PR_USE_SKIPNA', True) def test_compute_and_use_skipna_handling(self): """Test bucket resampler computation and use skipna handling.""" import dask.array as da data = da.ones((5,)) self._compute_mocked_bucket_sum(data, mask_all_nan=True) self.bucket.resampler.get_sum.assert_called_once_with( data, skipna=True) self._compute_mocked_bucket_sum(data, skipna=False) self.bucket.resampler.get_sum.assert_called_once_with( data, skipna=False) self._compute_mocked_bucket_sum(data) self.bucket.resampler.get_sum.assert_called_once_with( data, skipna=True)
[docs] @mock.patch('satpy.resample.PR_USE_SKIPNA', False) def test_compute_and_not_use_skipna_handling(self): """Test bucket resampler computation and not use skipna handling.""" import dask.array as da data = da.ones((5,)) self._compute_mocked_bucket_sum(data, mask_all_nan=True) self.bucket.resampler.get_sum.assert_called_once_with( data, mask_all_nan=True) self._compute_mocked_bucket_sum(data, mask_all_nan=False) self.bucket.resampler.get_sum.assert_called_once_with( data, mask_all_nan=False) self._compute_mocked_bucket_sum(data) self.bucket.resampler.get_sum.assert_called_once_with( data, mask_all_nan=False) self._compute_mocked_bucket_sum(data, fill_value=2, skipna=True) self.bucket.resampler.get_sum.assert_called_once_with( data, fill_value=2, mask_all_nan=False)
[docs]class TestBucketCount(unittest.TestCase): """Test the count bucket resampler."""
[docs] def setUp(self): """Create fake area definitions and resampler to be tested.""" from satpy.resample import BucketCount get_lonlats = mock.MagicMock() get_lonlats.return_value = (1, 2) self.source_geo_def = mock.MagicMock(get_lonlats=get_lonlats) self.target_geo_def = mock.MagicMock(get_lonlats=get_lonlats) self.bucket = BucketCount(self.source_geo_def, self.target_geo_def)
def _compute_mocked_bucket_count(self, data, return_data=None, **kwargs): """Compute the mocked bucket count.""" self.bucket.resampler = mock.MagicMock() if return_data is not None: self.bucket.resampler.get_count.return_value = return_data else: self.bucket.resampler.get_count.return_value = data res = self.bucket.compute(data, **kwargs) return res
[docs] def test_compute(self): """Test count bucket resampler computation.""" import dask.array as da # 1D data data = da.ones((5,)) res = self._compute_mocked_bucket_count(data) self.bucket.resampler.get_count.assert_called_once_with() self.assertEqual(res.shape, (1, 5)) # 2D data data = da.ones((5, 5)) res = self._compute_mocked_bucket_count(data) self.bucket.resampler.get_count.assert_called_once_with() self.assertEqual(res.shape, (1, 5, 5)) # 3D data data = da.ones((3, 5, 5)) res = self._compute_mocked_bucket_count(data, return_data=data[0, :, :]) self.assertEqual(res.shape, (3, 5, 5))
[docs]class TestBucketFraction(unittest.TestCase): """Test the fraction bucket resampler."""
[docs] def setUp(self): """Create fake area definitions and resampler to be tested.""" from satpy.resample import BucketFraction get_lonlats = mock.MagicMock() get_lonlats.return_value = (1, 2) get_proj_vectors = mock.MagicMock() get_proj_vectors.return_value = ([1, 2, 3, 4, 5], [1, 2, 3, 4, 5]) self.source_geo_def = mock.MagicMock(get_lonlats=get_lonlats) self.target_geo_def = mock.MagicMock(get_lonlats=get_lonlats, crs=None, get_proj_vectors=get_proj_vectors) self.bucket = BucketFraction(self.source_geo_def, self.target_geo_def)
[docs] def test_compute(self): """Test fraction bucket resampler computation.""" import dask.array as da import numpy as np self.bucket.resampler = mock.MagicMock() data = da.ones((3, 3)) # No kwargs given _ = self.bucket.compute(data) self.bucket.resampler.get_fractions.assert_called_with( data, categories=None, fill_value=np.nan) # Custom kwargs _ = self.bucket.compute(data, categories=[1, 2], fill_value=0) self.bucket.resampler.get_fractions.assert_called_with( data, categories=[1, 2], fill_value=0) # Too many dimensions data = da.ones((3, 5, 5)) with self.assertRaises(ValueError): _ = self.bucket.compute(data)
[docs] @mock.patch('pyresample.bucket.BucketResampler') def test_resample(self, pyresample_bucket): """Test fraction bucket resamplers resample method.""" import dask.array as da import numpy as np import xarray as xr self.bucket.resampler = mock.MagicMock() self.bucket.precompute = mock.MagicMock() self.bucket.compute = mock.MagicMock() # Fractions return a dict data = xr.DataArray(da.ones((1, 5, 5)), dims=('bands', 'y', 'x')) arr = da.ones((5, 5)) self.bucket.compute.return_value = {0: arr, 1: arr, 2: arr} res = self.bucket.resample(data) self.assertTrue('categories' in res.coords) self.assertTrue('categories' in res.dims) self.assertTrue(np.all(res.coords['categories'] == np.array([0, 1, 2])))