Source code for satpy.tests.reader_tests.test_avhrr_l1b_gaclac

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2009-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/>.
"""Pygac interface."""

from datetime import datetime
from unittest import TestCase, mock

import dask.array as da
import numpy as np
import pytest
import xarray as xr

GAC_PATTERN = '{creation_site:3s}.{transfer_mode:4s}.{platform_id:2s}.D{start_time:%y%j.S%H%M}.E{end_time:%H%M}.B{orbit_number:05d}{end_orbit_last_digits:02d}.{station:2s}'  # noqa

GAC_POD_FILENAMES = ["NSS.GHRR.NA.D79184.S1150.E1337.B0008384.WI",
                     "NSS.GHRR.NA.D79184.S2350.E0137.B0008384.WI",
                     "NSS.GHRR.NA.D80021.S0927.E1121.B0295354.WI",
                     "NSS.GHRR.NA.D80021.S1120.E1301.B0295455.WI",
                     "NSS.GHRR.NA.D80021.S1256.E1450.B0295556.GC",
                     "NSS.GHRR.NE.D83208.S1219.E1404.B0171819.WI",
                     "NSS.GHRR.NG.D88002.S0614.E0807.B0670506.WI",
                     "NSS.GHRR.TN.D79183.S1258.E1444.B0369697.GC",
                     "NSS.GHRR.TN.D80003.S1147.E1332.B0630506.GC",
                     "NSS.GHRR.TN.D80003.S1328.E1513.B0630507.GC",
                     "NSS.GHRR.TN.D80003.S1509.E1654.B0630608.GC"]

GAC_KLM_FILENAMES = ["NSS.GHRR.NK.D01235.S0252.E0446.B1703233.GC",
                     "NSS.GHRR.NL.D01288.S2315.E0104.B0549495.GC",
                     "NSS.GHRR.NM.D04111.S2305.E0050.B0947778.GC",
                     "NSS.GHRR.NN.D13011.S0559.E0741.B3939192.WI",
                     "NSS.GHRR.NP.D15361.S0121.E0315.B3547172.SV",
                     "NSS.GHRR.M1.D15362.S0031.E0129.B1699697.SV",
                     "NSS.GHRR.M2.D10178.S2359.E0142.B1914142.SV"]

LAC_POD_FILENAMES = ["BRN.HRPT.ND.D95152.S1730.E1715.B2102323.UB",
                     "BRN.HRPT.ND.D95152.S1910.E1857.B2102424.UB",
                     "BRN.HRPT.NF.D85152.S1345.E1330.B0241414.UB",
                     "BRN.HRPT.NJ.D95152.S1233.E1217.B0216060.UB"]

LAC_KLM_FILENAMES = ["BRN.HRPT.M1.D14152.S0958.E1012.B0883232.UB",
                     "BRN.HRPT.M1.D14152.S1943.E1958.B0883838.UB",
                     "BRN.HRPT.M2.D12153.S0912.E0922.B2914747.UB",
                     "BRN.HRPT.NN.D12153.S0138.E0152.B3622828.UB",
                     "BRN.HRPT.NN.D12153.S0139.E0153.B3622828.UB",
                     "BRN.HRPT.NN.D12153.S1309.E1324.B3623535.UB",
                     "BRN.HRPT.NP.D12153.S0003.E0016.B1707272.UB",
                     "BRN.HRPT.NP.D12153.S1134.E1148.B1707979.UB",
                     "BRN.HRPT.NP.D16184.S1256.E1311.B3813131.UB",
                     "BRN.HRPT.NP.D16184.S1438.E1451.B3813232.UB",
                     "BRN.HRPT.NP.D16184.S1439.E1451.B3813232.UB",
                     "BRN.HRPT.NP.D16185.S1245.E1259.B3814545.UB",
                     "BRN.HRPT.NP.D16185.S1427.E1440.B3814646.UB",
                     "NSS.FRAC.M2.D12153.S1729.E1910.B2915354.SV",
                     "NSS.LHRR.NP.D16306.S1803.E1814.B3985555.WI"]


[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile.__init__", return_value=None) def _get_fh_mocked(init_mock, **attrs): """Create a mocked file handler with the given attributes.""" from satpy.readers.avhrr_l1b_gaclac import GACLACFile fh = GACLACFile() for name, value in attrs.items(): setattr(fh, name, value) return fh
[docs] def _get_reader_mocked(along_track=3): """Create a mocked reader.""" reader = mock.MagicMock(spacecraft_name="spacecraft_name", meta_data={"foo": "bar"}) reader.mask = [0, 0] reader.get_times.return_value = np.arange(along_track) reader.get_tle_lines.return_value = "tle" return reader
[docs] class PygacPatcher(TestCase): """Patch pygac."""
[docs] def setUp(self): """Patch pygac imports.""" self.pygac = mock.MagicMock() self.fhs = mock.MagicMock() modules = { "pygac": self.pygac, "pygac.gac_klm": self.pygac.gac_klm, "pygac.gac_pod": self.pygac.gac_pod, "pygac.lac_klm": self.pygac.lac_klm, "pygac.lac_pod": self.pygac.lac_pod, "pygac.utils": self.pygac.utils, "pygac.calibration": self.pygac.calibration, } self.module_patcher = mock.patch.dict("sys.modules", modules) self.module_patcher.start()
[docs] def tearDown(self): """Unpatch the pygac imports.""" self.module_patcher.stop()
[docs] class GACLACFilePatcher(PygacPatcher): """Patch pygac."""
[docs] def setUp(self): """Patch GACLACFile.""" super().setUp() # Import GACLACFile here to make it patchable. Otherwise self._get_fh # might import it first which would prevent a successful patch. from satpy.readers.avhrr_l1b_gaclac import GACLACFile self.GACLACFile = GACLACFile
[docs] class TestGACLACFile(GACLACFilePatcher): """Test the GACLAC file handler."""
[docs] def _get_fh(self, filename="NSS.GHRR.NG.D88002.S0614.E0807.B0670506.WI", **kwargs): """Create a file handler.""" from trollsift import parse filename_info = parse(GAC_PATTERN, filename) return self.GACLACFile(filename, filename_info, {}, **kwargs)
[docs] def test_init(self): """Test GACLACFile initialization.""" from pygac.gac_klm import GACKLMReader from pygac.gac_pod import GACPODReader from pygac.lac_klm import LACKLMReader from pygac.lac_pod import LACPODReader kwargs = {"start_line": 1, "end_line": 2, "strip_invalid_coords": True, "interpolate_coords": True, "adjust_clock_drift": True, "tle_dir": "tle_dir", "tle_name": "tle_name", "tle_thresh": 123, "calibration": "calibration"} for filenames, reader_cls in zip([GAC_POD_FILENAMES, GAC_KLM_FILENAMES, LAC_POD_FILENAMES, LAC_KLM_FILENAMES], [GACPODReader, GACKLMReader, LACPODReader, LACKLMReader]): for filename in filenames: fh = self._get_fh(filename, **kwargs) assert fh.start_time < fh.end_time assert fh.reader_class is reader_cls
[docs] def test_read_raw_data(self): """Test raw data reading.""" fh = _get_fh_mocked(reader=None, interpolate_coords="interpolate_coords", creation_site="creation_site", reader_kwargs={"foo": "bar"}, filename="myfile") reader = mock.MagicMock(mask=[0]) reader_cls = mock.MagicMock(return_value=reader) fh.reader_class = reader_cls fh.read_raw_data() reader_cls.assert_called_with(interpolate_coords="interpolate_coords", creation_site="creation_site", foo="bar") reader.read.assert_called_with("myfile") # Test exception if all data is masked reader.mask = [1] fh.reader = None with pytest.raises(ValueError, match="All data is masked out"): fh.read_raw_data()
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._update_attrs") @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile.slice") @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._get_channel") def test_get_dataset_slice(self, get_channel, slc, *mocks): """Get a slice of a dataset.""" from satpy.tests.utils import make_dataid # Test slicing/stripping def slice_patched(data, times): if len(data.shape) == 2: return data[1:3, :], times[1:3] return data[1:3], times[1:3] ch = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12], [13, 14, 15]]) acq = np.array([0, 1, 2, 3, 4]) slc.side_effect = slice_patched get_channel.return_value = ch kwargs_list = [{"strip_invalid_coords": False, "start_line": 123, "end_line": 456}, {"strip_invalid_coords": True, "start_line": None, "end_line": None}, {"strip_invalid_coords": True, "start_line": 123, "end_line": 456}] for kwargs in kwargs_list: fh = _get_fh_mocked( reader=_get_reader_mocked(along_track=len(acq)), chn_dict={"1": 0}, **kwargs ) key = make_dataid(name="1", calibration="reflectance") info = {"name": "1", "standard_name": "reflectance"} res = fh.get_dataset(key, info) np.testing.assert_array_equal(res.data, ch[1:3, :]) np.testing.assert_array_equal(res.coords["acq_time"].data, acq[1:3]) np.testing.assert_array_equal(slc.call_args_list[-1][1]["times"], acq) np.testing.assert_array_equal(slc.call_args_list[-1][1]["data"], ch)
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._update_attrs") def test_get_dataset_latlon(self, *mocks): """Test getting the latitudes and longitudes.""" from satpy.tests.utils import make_dataid lons = np.ones((3, 3)) lats = 2 * lons reader = _get_reader_mocked() reader.get_lonlat.return_value = lons, lats fh = _get_fh_mocked( reader=reader, start_line=None, end_line=None, strip_invalid_coords=False, interpolate_coords=True ) # With interpolation of coordinates for name, exp_data in zip(["longitude", "latitude"], [lons, lats]): key = make_dataid(name=name) info = {"name": name, "standard_name": "my_standard_name"} res = fh.get_dataset(key=key, info=info) exp = xr.DataArray(exp_data, name=res.name, dims=("y", "x"), coords={"acq_time": ("y", [0, 1, 2])}) xr.testing.assert_equal(res, exp) # Without interpolation of coordinates fh.interpolate_coords = False for name, _exp_data in zip(["longitude", "latitude"], [lons, lats]): key = make_dataid(name=name) info = {"name": name, "standard_name": "my_standard_name"} res = fh.get_dataset(key=key, info=info) assert res.dims == ("y", "x_every_eighth")
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._update_attrs") @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._get_angle") def test_get_dataset_angles(self, get_angle, *mocks): """Test getting the angles.""" from satpy.readers.avhrr_l1b_gaclac import ANGLES from satpy.tests.utils import make_dataid ones = np.ones((3, 3)) get_angle.return_value = ones reader = _get_reader_mocked() fh = _get_fh_mocked( reader=reader, start_line=None, end_line=None, strip_invalid_coords=False, interpolate_coords=True ) # With interpolation of coordinates for angle in ANGLES: key = make_dataid(name=angle) info = {"name": angle, "standard_name": "my_standard_name"} res = fh.get_dataset(key=key, info=info) exp = xr.DataArray(ones, name=res.name, dims=("y", "x"), coords={"acq_time": ("y", [0, 1, 2])}) xr.testing.assert_equal(res, exp) # Without interpolation of coordinates fh.interpolate_coords = False for angle in ANGLES: key = make_dataid(name=angle) info = {"name": angle, "standard_name": "my_standard_name"} res = fh.get_dataset(key=key, info=info) assert res.dims == ("y", "x_every_eighth")
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._update_attrs") def test_get_dataset_qual_flags(self, *mocks): """Test getting the qualitiy flags.""" from satpy.tests.utils import make_dataid qual_flags = np.ones((3, 7)) reader = _get_reader_mocked() reader.get_qual_flags.return_value = qual_flags fh = _get_fh_mocked( reader=reader, start_line=None, end_line=None, strip_invalid_coords=False, interpolate_coords=True ) key = make_dataid(name="qual_flags") info = {"name": "qual_flags"} res = fh.get_dataset(key=key, info=info) exp = xr.DataArray(qual_flags, name=res.name, dims=("y", "num_flags"), coords={"acq_time": ("y", [0, 1, 2]), "num_flags": ["Scan line number", "Fatal error flag", "Insufficient data for calibration", "Insufficient data for calibration", "Solar contamination of blackbody in channels 3", "Solar contamination of blackbody in channels 4", "Solar contamination of blackbody in channels 5"]}) xr.testing.assert_equal(res, exp)
[docs] def test_get_channel(self): """Test getting the channels.""" from satpy.tests.utils import make_dataid counts = np.moveaxis(np.array([[[1, 2, 3], [4, 5, 6]]]), 0, 2) calib_channels = 2 * counts reader = _get_reader_mocked() reader.get_counts.return_value = counts reader.get_calibrated_channels.return_value = calib_channels fh = _get_fh_mocked(reader=reader, counts=None, calib_channels=None, chn_dict={"1": 0}) key = make_dataid(name="1", calibration="counts") # Counts res = fh._get_channel(key=key) np.testing.assert_array_equal(res, [[1, 2, 3], [4, 5, 6]]) np.testing.assert_array_equal(fh.counts, counts) # Reflectance and Brightness Temperature for calib in ["reflectance", "brightness_temperature"]: key = make_dataid(name="1", calibration=calib) res = fh._get_channel(key=key) np.testing.assert_array_equal(res, [[2, 4, 6], [8, 10, 12]]) np.testing.assert_array_equal(fh.calib_channels, calib_channels) # Invalid with pytest.raises(ValueError, match="coffee invalid value for <enum 'calibration'>"): _ = make_dataid(name="7", calibration="coffee") # Buffering reader.get_counts.reset_mock() key = make_dataid(name="1", calibration="counts") fh._get_channel(key=key) reader.get_counts.assert_not_called() reader.get_calibrated_channels.reset_mock() for calib in ["reflectance", "brightness_temperature"]: key = make_dataid(name="1", calibration=calib) fh._get_channel(key) reader.get_calibrated_channels.assert_not_called()
[docs] def test_get_angle(self): """Test getting the angle.""" from satpy.tests.utils import make_dataid reader = mock.MagicMock() reader.get_angles.return_value = 1, 2, 3, 4, 5 fh = _get_fh_mocked(reader=reader, angles=None) # Test angle readout key = make_dataid(name="sensor_zenith_angle") res = fh._get_angle(key) assert res == 2 assert fh.angles == {"sensor_zenith_angle": 2, "sensor_azimuth_angle": 1, "solar_zenith_angle": 4, "solar_azimuth_angle": 3, "sun_sensor_azimuth_difference_angle": 5} # Test buffering key = make_dataid(name="sensor_azimuth_angle") fh._get_angle(key) reader.get_angles.assert_called_once()
[docs] def test_strip_invalid_lat(self): """Test stripping invalid coordinates.""" import pygac.utils reader = mock.MagicMock() reader.get_lonlat.return_value = None, None fh = _get_fh_mocked(reader=reader, first_valid_lat=None) # Test stripping pygac.utils.strip_invalid_lat.return_value = 1, 2 start, end = fh._strip_invalid_lat() assert (start, end) == (1, 2) # Test buffering fh._strip_invalid_lat() pygac.utils.strip_invalid_lat.assert_called_once()
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._slice") def test_slice(self, _slice): # noqa: PT019 """Test slicing.""" def _slice_patched(data): return data[1:3] _slice.side_effect = _slice_patched data = np.zeros((4, 2)) times = np.array([1, 2, 3, 4], dtype="datetime64[us]") fh = _get_fh_mocked(start_line=1, end_line=3, strip_invalid_coords=False) data_slc, times_slc = fh.slice(data, times) np.testing.assert_array_equal(data_slc, data[1:3]) np.testing.assert_array_equal(times_slc, times[1:3]) assert fh.start_time == datetime(1970, 1, 1, 0, 0, 0, 2) assert fh.end_time == datetime(1970, 1, 1, 0, 0, 0, 3)
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._get_qual_flags") @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._strip_invalid_lat") def test__slice(self, strip_invalid_lat, get_qual_flags): """Test slicing.""" import pygac.utils pygac.utils.check_user_scanlines.return_value = 1, 2 pygac.utils.slice_channel.return_value = "sliced" strip_invalid_lat.return_value = 3, 4 get_qual_flags.return_value = "qual_flags" data = np.zeros((2, 2)) # a) Only start/end line given fh = _get_fh_mocked(start_line=5, end_line=6, strip_invalid_coords=False) data_slc = fh._slice(data) assert data_slc == "sliced" pygac.utils.check_user_scanlines.assert_called_with( start_line=5, end_line=6, first_valid_lat=None, last_valid_lat=None, along_track=2) pygac.utils.slice_channel.assert_called_with( data, start_line=1, end_line=2, first_valid_lat=None, last_valid_lat=None) # b) Only strip_invalid_coords=True fh = _get_fh_mocked(start_line=None, end_line=None, strip_invalid_coords=True) fh._slice(data) pygac.utils.check_user_scanlines.assert_called_with( start_line=0, end_line=0, first_valid_lat=3, last_valid_lat=4, along_track=2) # c) Both fh = _get_fh_mocked(start_line=5, end_line=6, strip_invalid_coords=True) fh._slice(data) pygac.utils.check_user_scanlines.assert_called_with( start_line=5, end_line=6, first_valid_lat=3, last_valid_lat=4, along_track=2) # Test slicing with older pygac versions pygac.utils.slice_channel.return_value = ("sliced", "foo", "bar") data_slc = fh._slice(data) assert data_slc == "sliced"
[docs] class TestGetDataset(GACLACFilePatcher): """Test the get_dataset method."""
[docs] def setUp(self): """Set up the instance.""" self.exp = xr.DataArray(da.ones((3, 3)), name="1", dims=("y", "x"), coords={"acq_time": ("y", [0, 1, 2])}, attrs={"name": "1", "platform_name": "spacecraft_name", "orbit_number": 123, "sensor": "sensor", "foo": "bar", "standard_name": "my_standard_name"}) self.exp.coords["acq_time"].attrs["long_name"] = "Mean scanline acquisition time" super().setUp()
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile.__init__", return_value=None) @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile.read_raw_data") @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._get_channel", return_value=np.ones((3, 3))) def test_get_dataset_channels(self, get_channel, *mocks): """Test getting the channel datasets.""" pygac_reader = _get_reader_mocked() fh = self._create_file_handler(pygac_reader) # Test calibration to reflectance as well as attributes. key, res = self._get_dataset(fh) exp = self._create_expected(res.name) exp.attrs["orbital_parameters"] = {"tle": "tle"} xr.testing.assert_identical(res, exp) get_channel.assert_called_with(key) self._check_get_channel_calls(fh, get_channel)
[docs] @staticmethod def _get_dataset(fh): from satpy.tests.utils import make_dataid key = make_dataid(name="1", calibration="reflectance") info = {"name": "1", "standard_name": "my_standard_name"} res = fh.get_dataset(key=key, info=info) return key, res
[docs] @staticmethod def _create_file_handler(reader): """Mock reader and file handler.""" fh = _get_fh_mocked( reader=reader, chn_dict={"1": 0, "5": 0}, start_line=None, end_line=None, strip_invalid_coords=False, filename_info={"orbit_number": 123}, sensor="sensor", ) return fh
[docs] @staticmethod def _create_expected(name): exp = xr.DataArray(da.ones((3, 3)), name=name, dims=("y", "x"), coords={"acq_time": ("y", [0, 1, 2])}, attrs={"name": "1", "platform_name": "spacecraft_name", "orbit_number": 123, "sensor": "sensor", "foo": "bar", "standard_name": "my_standard_name"}) exp.coords["acq_time"].attrs["long_name"] = "Mean scanline acquisition time" return exp
[docs] @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile.__init__", return_value=None) @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile.read_raw_data") @mock.patch("satpy.readers.avhrr_l1b_gaclac.GACLACFile._get_channel", return_value=np.ones((3, 3))) def test_get_dataset_no_tle(self, get_channel, *mocks): """Test getting the channel datasets when no TLEs are present.""" pygac_reader = _get_reader_mocked() pygac_reader.get_tle_lines = mock.MagicMock() pygac_reader.get_tle_lines.side_effect = RuntimeError() fh = self._create_file_handler(pygac_reader) # Test calibration to reflectance as well as attributes. key, res = self._get_dataset(fh) exp = self._create_expected(res.name) xr.testing.assert_identical(res, exp) get_channel.assert_called_with(key) self._check_get_channel_calls(fh, get_channel)
[docs] @staticmethod def _check_get_channel_calls(fh, get_channel): """Check _get_channel() calls.""" from satpy.tests.utils import make_dataid for key in [make_dataid(name="1", calibration="counts"), make_dataid(name="5", calibration="brightness_temperature")]: fh.get_dataset(key=key, info={"name": 1}) get_channel.assert_called_with(key)