#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2019, 2022 Satpy developers
#
# This file is part of satpy.
#
# satpy is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# satpy is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# satpy. If not, see <http://www.gnu.org/licenses/>.
"""Test the eps l1b format."""
import os
from contextlib import suppress
from tempfile import mkstemp
from unittest import TestCase, mock
import numpy as np
import pytest
import xarray as xr
import satpy
from satpy._config import get_config_path
from satpy.readers import eps_l1b as eps
from satpy.tests.utils import make_dataid
# NOTE:
# The following fixtures are not defined in this file, but are used and injected by Pytest:
# - caplog
grh_dtype = np.dtype([("record_class", "|i1"),
("INSTRUMENT_GROUP", "|i1"),
("RECORD_SUBCLASS", "|i1"),
("RECORD_SUBCLASS_VERSION", "|i1"),
("RECORD_SIZE", ">u4"),
("RECORD_START_TIME", "S6"),
("RECORD_STOP_TIME", "S6")])
[docs]
def create_sections(structure):
"""Create file sections."""
sections = {}
format_fn = get_config_path("eps_avhrrl1b_6.5.xml")
form = eps.XMLFormat(format_fn)
for count, (rec_class, sub_class) in structure:
try:
the_dtype = form.dtype((rec_class, sub_class))
except KeyError:
continue
item_size = the_dtype.itemsize + grh_dtype.itemsize
the_dtype = np.dtype(grh_dtype.descr + the_dtype.descr)
item = np.zeros(count, the_dtype)
item["record_class"] = eps.record_class.index(rec_class)
item["RECORD_SUBCLASS"] = sub_class
item["RECORD_SIZE"] = item_size
sections[(rec_class, sub_class)] = item
return sections
[docs]
class BaseTestCaseEPSL1B(TestCase):
"""Base class for EPS l1b test case."""
[docs]
def _create_structure(self):
structure = [(1, ("mphr", 0)), (1, ("sphr", 0)), (11, ("ipr", 0)),
(1, ("geadr", 1)), (1, ("geadr", 2)), (1, ("geadr", 3)),
(1, ("geadr", 4)), (1, ("geadr", 5)), (1, ("geadr", 6)),
(1, ("geadr", 7)), (1, ("giadr", 1)), (1, ("giadr", 2)),
(1, ("veadr", 1)), (self.scan_lines, ("mdr", 2))]
sections = create_sections(structure)
return sections
[docs]
class TestEPSL1B(BaseTestCaseEPSL1B):
"""Test the filehandler."""
[docs]
def setUp(self):
"""Set up the tests."""
# ipr is not present in the xml format ?
self.scan_lines = 1080
self.earth_views = 2048
sections = self._create_structure()
sections[("mphr", 0)]["TOTAL_MDR"] = (b"TOTAL_MDR = " +
bytes(str(self.scan_lines), encoding="ascii") +
b"\n")
sections[("mphr", 0)]["SPACECRAFT_ID"] = b"SPACECRAFT_ID = M03\n"
sections[("mphr", 0)]["INSTRUMENT_ID"] = b"INSTRUMENT_ID = AVHR\n"
sections[("sphr", 0)]["EARTH_VIEWS_PER_SCANLINE"] = (b"EARTH_VIEWS_PER_SCANLINE = " +
bytes(str(self.earth_views), encoding="ascii") +
b"\n")
sections[("sphr", 0)]["NAV_SAMPLE_RATE"] = b"NAV_SAMPLE_RATE = 20\n"
_fd, fname = mkstemp()
fd = open(_fd)
self.filename = fname
for _, arr in sections.items():
arr.tofile(fd)
fd.close()
self.fh = eps.EPSAVHRRFile(self.filename, {"start_time": "now",
"end_time": "later"}, {})
[docs]
def test_read_all(self):
"""Test initialization."""
self.fh._read_all()
assert self.fh.scanlines == 1080
assert self.fh.pixels == 2048
[docs]
def test_dataset(self):
"""Test getting a dataset."""
did = make_dataid(name="1", calibration="reflectance")
res = self.fh.get_dataset(did, {})
assert isinstance(res, xr.DataArray)
assert res.attrs["platform_name"] == "Metop-C"
assert res.attrs["sensor"] == "avhrr-3"
assert res.attrs["name"] == "1"
assert res.attrs["calibration"] == "reflectance"
assert res.attrs["units"] == "%"
did = make_dataid(name="4", calibration="brightness_temperature")
res = self.fh.get_dataset(did, {})
assert isinstance(res, xr.DataArray)
assert res.attrs["platform_name"] == "Metop-C"
assert res.attrs["sensor"] == "avhrr-3"
assert res.attrs["name"] == "4"
assert res.attrs["calibration"] == "brightness_temperature"
assert res.attrs["units"] == "K"
[docs]
def test_get_dataset_radiance(self):
"""Test loading a data array with radiance calibration."""
did = make_dataid(name="1", calibration="radiance")
res = self.fh.get_dataset(did, {})
assert isinstance(res, xr.DataArray)
assert res.attrs["platform_name"] == "Metop-C"
assert res.attrs["sensor"] == "avhrr-3"
assert res.attrs["name"] == "1"
assert res.attrs["calibration"] == "radiance"
assert res.attrs["units"] == "W m^-2 sr^-1"
[docs]
def test_navigation(self):
"""Test the navigation."""
did = make_dataid(name="longitude")
res = self.fh.get_dataset(did, {})
assert isinstance(res, xr.DataArray)
assert res.attrs["platform_name"] == "Metop-C"
assert res.attrs["sensor"] == "avhrr-3"
assert res.attrs["name"] == "longitude"
[docs]
def test_angles(self):
"""Test the navigation."""
did = make_dataid(name="solar_zenith_angle")
res = self.fh.get_dataset(did, {})
assert isinstance(res, xr.DataArray)
assert res.attrs["platform_name"] == "Metop-C"
assert res.attrs["sensor"] == "avhrr-3"
assert res.attrs["name"] == "solar_zenith_angle"
[docs]
def test_clould_flags(self):
"""Test getting the cloud flags."""
did = make_dataid(name="cloud_flags")
res = self.fh.get_dataset(did, {})
assert isinstance(res, xr.DataArray)
assert res.attrs["platform_name"] == "Metop-C"
assert res.attrs["sensor"] == "avhrr-3"
assert res.attrs["name"] == "cloud_flags"
[docs]
@mock.patch("satpy.readers.eps_l1b.EPSAVHRRFile.__getitem__")
def test_get_full_angles_twice(self, mock__getitem__):
"""Test get full angles twice."""
geotiemock = mock.Mock()
metop20kmto1km = geotiemock.metop20kmto1km
metop20kmto1km.side_effect = lambda x, y: (x.copy(), y.copy())
def mock_getitem(key):
data = {"ANGULAR_RELATIONS_FIRST": np.zeros((7, 4)),
"ANGULAR_RELATIONS": np.zeros((7, 103, 4)),
"ANGULAR_RELATIONS_LAST": np.zeros((7, 4)),
"NAV_SAMPLE_RATE": 20}
return data[key]
mock__getitem__.side_effect = mock_getitem
avhrr_reader = satpy.readers.eps_l1b.EPSAVHRRFile(
filename="foo",
filename_info={"start_time": "foo", "end_time": "bar"},
filetype_info={"foo": "bar"}
)
avhrr_reader.scanlines = 7
avhrr_reader.pixels = 2048
with mock.patch.dict("sys.modules", geotiepoints=geotiemock):
# Get dask arrays
sun_azi, sun_zen, sat_azi, sat_zen = avhrr_reader.get_full_angles()
# Convert to numpy array
sun_zen_np1 = np.array(sun_zen)
# Convert to numpy array again
sun_zen_np2 = np.array(sun_zen)
assert np.allclose(sun_zen_np1, sun_zen_np2)
[docs]
class TestWrongScanlinesEPSL1B(BaseTestCaseEPSL1B):
"""Test the filehandler on a corrupt file."""
[docs]
@pytest.fixture(autouse=True)
def _inject_fixtures(self, caplog):
"""Inject caplog."""
self._caplog = caplog
[docs]
def setUp(self):
"""Set up the tests."""
# ipr is not present in the xml format ?
self.scan_lines = 1080
self.earth_views = 2048
sections = self._create_structure()
sections[("mphr", 0)]["TOTAL_MDR"] = (b"TOTAL_MDR = " +
bytes(str(self.scan_lines - 2), encoding="ascii") +
b"\n")
sections[("mphr", 0)]["SPACECRAFT_ID"] = b"SPACECRAFT_ID = M03\n"
sections[("mphr", 0)]["INSTRUMENT_ID"] = b"INSTRUMENT_ID = AVHR\n"
sections[("sphr", 0)]["EARTH_VIEWS_PER_SCANLINE"] = (b"EARTH_VIEWS_PER_SCANLINE = " +
bytes(str(self.earth_views), encoding="ascii") +
b"\n")
sections[("sphr", 0)]["NAV_SAMPLE_RATE"] = b"NAV_SAMPLE_RATE = 20\n"
_fd, fname = mkstemp()
fd = open(_fd)
self.filename = fname
for _, arr in sections.items():
arr.tofile(fd)
fd.close()
self.fh = eps.EPSAVHRRFile(self.filename, {"start_time": "now",
"end_time": "later"}, {})
[docs]
def test_read_all_return_right_number_of_scan_lines(self):
"""Test scanline assignment."""
self.fh._read_all()
assert self.fh.scanlines == self.scan_lines
[docs]
def test_read_all_warns_about_scan_lines(self):
"""Test scanline assignment."""
self.fh._read_all()
assert "scanlines" in self._caplog.records[0].message
assert self._caplog.records[0].levelname == "WARNING"
[docs]
def test_read_all_assigns_int_scan_lines(self):
"""Test scanline assignment."""
self.fh._read_all()
assert isinstance(self.fh.scanlines, int)
[docs]
def test_get_dataset_longitude_shape_is_right(self):
"""Test that the shape of longitude is 1080."""
key = make_dataid(name="longitude")
longitudes = self.fh.get_dataset(key, dict())
assert longitudes.shape == (self.scan_lines, self.earth_views)
[docs]
def tearDown(self):
"""Tear down the tests."""
with suppress(OSError):
os.remove(self.filename)
[docs]
class TestWrongSamplingEPSL1B(BaseTestCaseEPSL1B):
"""Test the filehandler on a corrupt file."""
[docs]
@pytest.fixture(autouse=True)
def _inject_fixtures(self, caplog):
"""Inject caplog."""
self._caplog = caplog
[docs]
def setUp(self):
"""Set up the tests."""
self.scan_lines = 1080
self.earth_views = 2048
self.sample_rate = 23
sections = self._create_structure()
sections[("mphr", 0)]["TOTAL_MDR"] = (b"TOTAL_MDR = " +
bytes(str(self.scan_lines), encoding="ascii") +
b"\n")
sections[("mphr", 0)]["SPACECRAFT_ID"] = b"SPACECRAFT_ID = M03\n"
sections[("mphr", 0)]["INSTRUMENT_ID"] = b"INSTRUMENT_ID = AVHR\n"
sections[("sphr", 0)]["EARTH_VIEWS_PER_SCANLINE"] = (b"EARTH_VIEWS_PER_SCANLINE = " +
bytes(str(self.earth_views), encoding="ascii") +
b"\n")
sections[("sphr", 0)]["NAV_SAMPLE_RATE"] = (b"NAV_SAMPLE_RATE = " +
bytes(str(self.sample_rate), encoding="ascii") +
b"\n")
_fd, fname = mkstemp()
fd = open(_fd)
self.filename = fname
for _, arr in sections.items():
arr.tofile(fd)
fd.close()
self.fh = eps.EPSAVHRRFile(self.filename, {"start_time": "now",
"end_time": "later"}, {})
[docs]
def test_get_dataset_fails_because_of_wrong_sample_rate(self):
"""Test that lons fail to be interpolate."""
key = make_dataid(name="longitude")
with pytest.raises(NotImplementedError):
self.fh.get_dataset(key, dict())