#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2016-2018 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/>.
"""Module for testing the satpy.readers.olci_nc module."""
import datetime
import unittest
import unittest.mock as mock
[docs]
class TestOLCIReader(unittest.TestCase):
"""Test various olci_nc filehandlers."""
[docs]
@mock.patch("xarray.open_dataset")
def test_instantiate(self, mocked_dataset):
"""Test initialization of file handlers."""
import xarray as xr
from satpy.readers.olci_nc import NCOLCI1B, NCOLCI2, NCOLCIBase, NCOLCICal, NCOLCIChannelBase, NCOLCIGeo
from satpy.tests.utils import make_dataid
cal_data = xr.Dataset(
{
"solar_flux": (("bands"), [0, 1, 2]),
"detector_index": (("bands"), [0, 1, 2]),
},
{"bands": [0, 1, 2], },
)
ds_id = make_dataid(name="Oa01", calibration="reflectance")
ds_id2 = make_dataid(name="wsqf", calibration="reflectance")
filename_info = {"mission_id": "S3A", "dataset_name": "Oa01", "start_time": 0, "end_time": 0}
test = NCOLCIBase("somedir/somefile.nc", filename_info, "c")
test.get_dataset(ds_id, filename_info)
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
test = NCOLCICal("somedir/somefile.nc", filename_info, "c")
test.get_dataset(ds_id, filename_info)
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
test = NCOLCIGeo("somedir/somefile.nc", filename_info, "c")
test.get_dataset(ds_id, filename_info)
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
test = NCOLCIChannelBase("somedir/somefile.nc", filename_info, "c")
test.get_dataset(ds_id, filename_info)
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
cal = mock.Mock()
cal.nc = cal_data
test = NCOLCI1B("somedir/somefile.nc", filename_info, "c", cal)
test.get_dataset(ds_id, filename_info)
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
test = NCOLCI2("somedir/somefile.nc", filename_info, "c")
test.get_dataset(ds_id, {"nc_key": "the_key"})
test.get_dataset(ds_id2, {"nc_key": "the_key"})
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
[docs]
@mock.patch("xarray.open_dataset")
def test_open_file_objects(self, mocked_open_dataset):
"""Test initialization of file handlers."""
from satpy.readers.olci_nc import NCOLCIBase
filename_info = {"mission_id": "S3A", "dataset_name": "Oa01", "start_time": 0, "end_time": 0}
open_file = mock.MagicMock()
file_handler = NCOLCIBase(open_file, filename_info, "c")
# deepcode ignore W0104: This is a property that is actually a function call.
file_handler.nc # pylint: disable=W0104
mocked_open_dataset.assert_called()
open_file.open.assert_called()
assert (open_file.open.return_value in mocked_open_dataset.call_args[0] or
open_file.open.return_value == mocked_open_dataset.call_args[1].get("filename_or_obj"))
[docs]
@mock.patch("xarray.open_dataset")
def test_get_mask(self, mocked_dataset):
"""Test reading datasets."""
import numpy as np
import xarray as xr
from satpy.readers.olci_nc import NCOLCI2
from satpy.tests.utils import make_dataid
mocked_dataset.return_value = xr.Dataset({"mask": (["rows", "columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6))},
coords={"rows": np.arange(5),
"columns": np.arange(6)})
ds_id = make_dataid(name="mask")
filename_info = {"mission_id": "S3A", "dataset_name": "mask", "start_time": 0, "end_time": 0}
test = NCOLCI2("somedir/somefile.nc", filename_info, "c")
res = test.get_dataset(ds_id, {"nc_key": "mask"})
assert res.dtype == np.dtype("bool")
expected = np.array([[True, False, True, True, True, True],
[False, False, True, True, False, False],
[False, False, False, False, False, True],
[False, True, False, False, False, True],
[True, False, False, True, False, False]])
np.testing.assert_array_equal(res.values, expected)
[docs]
@mock.patch("xarray.open_dataset")
def test_get_mask_with_alternative_items(self, mocked_dataset):
"""Test reading datasets."""
import numpy as np
import xarray as xr
from satpy.readers.olci_nc import NCOLCI2
from satpy.tests.utils import make_dataid
mocked_dataset.return_value = xr.Dataset({"mask": (["rows", "columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6))},
coords={"rows": np.arange(5),
"columns": np.arange(6)})
ds_id = make_dataid(name="mask")
filename_info = {"mission_id": "S3A", "dataset_name": "mask", "start_time": 0, "end_time": 0}
test = NCOLCI2("somedir/somefile.nc", filename_info, "c", mask_items=["INVALID"])
res = test.get_dataset(ds_id, {"nc_key": "mask"})
assert res.dtype == np.dtype("bool")
expected = np.array([True] + [False] * 29).reshape(5, 6)
np.testing.assert_array_equal(res.values, expected)
[docs]
@mock.patch("xarray.open_dataset")
def test_olci_angles(self, mocked_dataset):
"""Test reading datasets."""
import numpy as np
import xarray as xr
from satpy.readers.olci_nc import NCOLCIAngles
from satpy.tests.utils import make_dataid
attr_dict = {
"ac_subsampling_factor": 1,
"al_subsampling_factor": 2,
}
mocked_dataset.return_value = xr.Dataset({"SAA": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6)),
"SZA": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6)),
"OAA": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6)),
"OZA": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6))},
coords={"rows": np.arange(5),
"columns": np.arange(6)},
attrs=attr_dict)
filename_info = {"mission_id": "S3A", "dataset_name": "Oa01", "start_time": 0, "end_time": 0}
ds_id = make_dataid(name="solar_azimuth_angle")
ds_id2 = make_dataid(name="satellite_zenith_angle")
test = NCOLCIAngles("somedir/somefile.nc", filename_info, "c")
test.get_dataset(ds_id, filename_info)
test.get_dataset(ds_id2, filename_info)
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
[docs]
@mock.patch("xarray.open_dataset")
def test_olci_meteo(self, mocked_dataset):
"""Test reading datasets."""
import numpy as np
import xarray as xr
from satpy.readers.olci_nc import NCOLCIMeteo
from satpy.tests.utils import make_dataid
attr_dict = {
"ac_subsampling_factor": 1,
"al_subsampling_factor": 2,
}
data = {"humidity": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6)),
"total_ozone": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6)),
"sea_level_pressure": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6)),
"total_columnar_water_vapour": (["tie_rows", "tie_columns"],
np.array([1 << x for x in range(30)]).reshape(5, 6))}
mocked_dataset.return_value = xr.Dataset(data,
coords={"rows": np.arange(5),
"columns": np.arange(6)},
attrs=attr_dict)
filename_info = {"mission_id": "S3A", "dataset_name": "humidity", "start_time": 0, "end_time": 0}
ds_id = make_dataid(name="humidity")
ds_id2 = make_dataid(name="total_ozone")
test = NCOLCIMeteo("somedir/somefile.nc", filename_info, "c")
test.get_dataset(ds_id, filename_info)
test.get_dataset(ds_id2, filename_info)
mocked_dataset.assert_called()
mocked_dataset.reset_mock()
[docs]
@mock.patch("xarray.open_dataset")
def test_chl_nn(self, mocked_dataset):
"""Test unlogging the chl_nn product."""
import numpy as np
import xarray as xr
from satpy.readers.olci_nc import NCOLCI2
from satpy.tests.utils import make_dataid
attr_dict = {
"ac_subsampling_factor": 64,
"al_subsampling_factor": 1,
}
data = {"CHL_NN": (["rows", "columns"],
np.arange(30).reshape(5, 6).astype(float),
{"units": "lg(re mg.m-3)"})}
mocked_dataset.return_value = xr.Dataset(data,
coords={"rows": np.arange(5),
"columns": np.arange(6)},
attrs=attr_dict)
ds_info = {"name": "chl_nn", "sensor": "olci", "resolution": 300,
"standard_name": "algal_pigment_concentration", "units": "lg(re mg.m-3)",
"coordinates": ("longitude", "latitude"), "file_type": "esa_l2_chl_nn", "nc_key": "CHL_NN",
"modifiers": ()}
filename_info = {"mission_id": "S3A", "datatype_id": "WFR",
"start_time": datetime.datetime(2019, 9, 24, 9, 29, 39),
"end_time": datetime.datetime(2019, 9, 24, 9, 32, 39),
"creation_time": datetime.datetime(2019, 9, 24, 11, 40, 26), "duration": 179, "cycle": 49,
"relative_orbit": 307, "frame": 1800, "centre": "MAR", "mode": "O", "timeliness": "NR",
"collection": "002"}
ds_id = make_dataid(name="chl_nn")
file_handler = NCOLCI2("somedir/somefile.nc", filename_info, None, unlog=True)
res = file_handler.get_dataset(ds_id, ds_info)
assert res.attrs["units"] == "mg.m-3"
assert res.values[-1, -1] == 1e29
[docs]
class TestBitFlags(unittest.TestCase):
"""Test the bitflag reading."""
[docs]
def test_bitflags(self):
"""Test the BitFlags class."""
from functools import reduce
import numpy as np
from satpy.readers.olci_nc import BitFlags
flag_list = ["INVALID", "WATER", "LAND", "CLOUD", "SNOW_ICE",
"INLAND_WATER", "TIDAL", "COSMETIC", "SUSPECT", "HISOLZEN",
"SATURATED", "MEGLINT", "HIGHGLINT", "WHITECAPS",
"ADJAC", "WV_FAIL", "PAR_FAIL", "AC_FAIL", "OC4ME_FAIL",
"OCNN_FAIL", "Extra_1", "KDM_FAIL", "Extra_2",
"CLOUD_AMBIGUOUS", "CLOUD_MARGIN", "BPAC_ON",
"WHITE_SCATT", "LOWRW", "HIGHRW"]
bits = np.array([1 << x for x in range(len(flag_list))])
bflags = BitFlags(bits)
items = ["INVALID", "SNOW_ICE", "INLAND_WATER", "SUSPECT",
"AC_FAIL", "CLOUD", "HISOLZEN", "OCNN_FAIL",
"CLOUD_MARGIN", "CLOUD_AMBIGUOUS", "LOWRW", "LAND"]
mask = reduce(np.logical_or, [bflags[item] for item in items])
expected = np.array([True, False, True, True, True, True, False,
False, True, True, False, False, False, False,
False, False, False, True, False, True, False,
False, False, True, True, False, False, True,
False])
assert all(mask == expected)