#!/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])))