Source code for satpy.tests.reader_tests.test_ascat_l2_soilmoisture_bufr

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2017-2021 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/>.
"""Unittesting the ASCAT SCATTEROMETER SOIL MOISTURE BUFR reader."""

import os
import sys
import unittest
from datetime import datetime

import numpy as np

# TDB: this test is based on test_seviri_l2_bufr.py and test_iasi_l2.py

# This is a test for ASCAT SoilMoisture product message, take from a real
# bufr file distributed over EUMETCAST


[docs] def create_message(): """Create fake message for testing.""" nlat = 10 nlon = 10 samples = nlat*nlon lat, lon = np.meshgrid(np.linspace(63, 65, nlat), np.linspace(-30, -20, nlon)) lat = np.round(np.ravel(lat), 4) lon = np.round(np.ravel(lon), 4) rstate = np.random.RandomState(0) surfaceSoilMoisture = np.round(rstate.rand(samples)*100, 1) surfaceSoilMoisture[0] = -1e+100 retmsg = { 'inputDelayedDescriptorReplicationFactor': [8], 'edition': 4, 'masterTableNumber': 0, 'bufrHeaderCentre': 254, 'bufrHeaderSubCentre': 0, 'updateSequenceNumber': 0, 'dataCategory': 12, 'internationalDataSubCategory': 255, 'dataSubCategory': 190, 'masterTablesVersionNumber': 13, 'localTablesVersionNumber': 0, 'typicalYear': 2020, 'typicalMonth': 12, 'typicalDay': 21, 'typicalHour': 9, 'typicalMinute': 33, 'typicalSecond': 0, 'numberOfSubsets': samples, 'observedData': 1, 'compressedData': 1, 'unexpandedDescriptors': 312061, 'centre': 254, 'subCentre': 0, '#1#softwareIdentification': 1000, 'satelliteIdentifier': 4, 'satelliteInstruments': 190, 'year': 2020, 'month': 12, 'day': 21, 'hour': 9, 'minute': 33, 'second': np.linspace(0, 59, samples), 'latitude': lat, 'longitude': lon, 'surfaceSoilMoisture': surfaceSoilMoisture, 'soilMoistureQuality': np.zeros(samples), } return retmsg
MSG = create_message() # the notional filename that would contain the above test message data FILENAME = 'W_XX-EUMETSAT-TEST,SOUNDING+SATELLITE,METOPA+ASCAT_C_EUMC_20201221093300_73545_eps_o_125_ssm_l2.bin' # the information that would be extracted from the above filename according to the pattern in the .yaml FILENAME_INFO = { 'reception_location': 'TEST', 'platform': 'METOPA', 'instrument': 'ASCAT', 'start_time': '20201221093300', 'perigee': '73545', 'species': '125_ssm', 'level': 'l2' } # file type info for the above file that is defined in the .yaml FILETYPE_INFO = { 'file_type': 'ascat_l2_soilmoisture_bufr', 'file_reader': 'AscatSoilMoistureBufr' }
[docs] def save_test_data(path): """Save the test file to the indicated directory.""" import eccodes as ec filepath = os.path.join(path, FILENAME) with open(filepath, "wb") as f: for m in [MSG]: buf = ec.codes_bufr_new_from_samples('BUFR4_local_satellite') for key in m: val = m[key] if np.isscalar(val): ec.codes_set(buf, key, val) else: ec.codes_set_array(buf, key, val) ec.codes_set(buf, 'pack', 1) ec.codes_write(buf, f) ec.codes_release(buf) return filepath
[docs] class TesitAscatL2SoilmoistureBufr(unittest.TestCase): """Test ASCAT Soil Mosture loader."""
[docs] def setUp(self): """Create temporary file to perform tests with.""" import tempfile from satpy.readers.ascat_l2_soilmoisture_bufr import AscatSoilMoistureBufr self.base_dir = tempfile.mkdtemp() self.fname = save_test_data(self.base_dir) self.fname_info = FILENAME_INFO self.ftype_info = FILETYPE_INFO self.reader = AscatSoilMoistureBufr(self.fname, self.fname_info, self.ftype_info)
[docs] def tearDown(self): """Remove the temporary directory created for a test.""" try: import shutil shutil.rmtree(self.base_dir, ignore_errors=True) except OSError: pass
[docs] @unittest.skipIf(sys.platform.startswith('win'), "'eccodes' not supported on Windows") def test_scene(self): """Test scene creation.""" from satpy import Scene fname = os.path.join(self.base_dir, FILENAME) scn = Scene(reader='ascat_l2_soilmoisture_bufr', filenames=[fname]) self.assertTrue('scatterometer' in scn.sensor_names) self.assertTrue(datetime(2020, 12, 21, 9, 33, 0) == scn.start_time) self.assertTrue(datetime(2020, 12, 21, 9, 33, 59) == scn.end_time)
[docs] @unittest.skipIf(sys.platform.startswith('win'), "'eccodes' not supported on Windows") def test_scene_load_available_datasets(self): """Test that all datasets are available.""" from satpy import Scene fname = os.path.join(self.base_dir, FILENAME) scn = Scene(reader='ascat_l2_soilmoisture_bufr', filenames=[fname]) self.assertTrue('surface_soil_moisture' in scn.available_dataset_names()) scn.load(scn.available_dataset_names()) loaded = [dataset.name for dataset in scn] self.assertTrue(sorted(loaded) == sorted(scn.available_dataset_names()))
[docs] @unittest.skipIf(sys.platform.startswith('win'), "'eccodes' not supported on Windows") def test_scene_dataset_values(self): """Test loading data.""" from satpy import Scene fname = os.path.join(self.base_dir, FILENAME) scn = Scene(reader='ascat_l2_soilmoisture_bufr', filenames=[fname]) for name in scn.available_dataset_names(): scn.load([name]) loaded_values = scn[name].values fill_value = scn[name].attrs['fill_value'] # replace nans in data loaded from file with the fill value defined in the .yaml # to make them comparable loaded_values_nan_filled = np.nan_to_num(loaded_values, nan=fill_value) key = scn[name].attrs['key'] original_values = MSG[key] # this makes each assertion below a separate test from unittest's point of view # (note: if all subtests pass, they will count as one test) with self.subTest(msg="Test failed for dataset: "+name): self.assertTrue(np.allclose(original_values, loaded_values_nan_filled))