#!/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])
assert "scatterometer" in scn.sensor_names
assert datetime(2020, 12, 21, 9, 33, 0) == scn.start_time
assert 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])
assert "surface_soil_moisture" in scn.available_dataset_names()
scn.load(scn.available_dataset_names())
loaded = [dataset.name for dataset in scn]
assert 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):
assert np.allclose(original_values, loaded_values_nan_filled)