Source code for satpy.tests.reader_tests.test_ahi_l1b_gridded_bin

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2020 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/>.
"""The ahi_l1b_gridded_bin reader tests package."""

import os
import shutil
import tempfile
import unittest
from unittest import mock

import dask.array as da
import numpy as np
from pyresample.geometry import AreaDefinition

from satpy.readers.ahi_l1b_gridded_bin import AHI_LUT_NAMES, AHIGriddedFileHandler


[docs] class TestAHIGriddedArea(unittest.TestCase): """Test the AHI gridded reader definition."""
[docs] def setUp(self): """Create fake data for testing.""" self.FULLDISK_SIZES = {0.005: {'x_size': 24000, 'y_size': 24000}, 0.01: {'x_size': 12000, 'y_size': 12000}, 0.02: {'x_size': 6000, 'y_size': 6000}} self.AHI_FULLDISK_EXTENT = [85., -60., 205., 60.]
[docs] @staticmethod def make_fh(filetype, area='fld'): """Create a test file handler.""" m = mock.mock_open() with mock.patch('satpy.readers.ahi_l1b_gridded_bin.open', m, create=True): fh = AHIGriddedFileHandler('somefile', {'area': area}, filetype_info={'file_type': filetype}) return fh
[docs] def test_low_res(self): """Check size of the low resolution (2km) grid.""" tmp_fh = self.make_fh('tir.01') self.assertEqual(self.FULLDISK_SIZES[0.02]['x_size'], tmp_fh.ncols) self.assertEqual(self.FULLDISK_SIZES[0.02]['y_size'], tmp_fh.nlines)
[docs] def test_med_res(self): """Check size of the low resolution (1km) grid.""" tmp_fh = self.make_fh('vis.02') self.assertEqual(self.FULLDISK_SIZES[0.01]['x_size'], tmp_fh.ncols) self.assertEqual(self.FULLDISK_SIZES[0.01]['y_size'], tmp_fh.nlines)
[docs] def test_hi_res(self): """Check size of the low resolution (0.5km) grid.""" tmp_fh = self.make_fh('ext.01') self.assertEqual(self.FULLDISK_SIZES[0.005]['x_size'], tmp_fh.ncols) self.assertEqual(self.FULLDISK_SIZES[0.005]['y_size'], tmp_fh.nlines)
[docs] def test_area_def(self): """Check that a valid full disk area is produced.""" good_area = AreaDefinition('gridded_himawari', 'A gridded Himawari area', 'longlat', 'EPSG:4326', self.FULLDISK_SIZES[0.01]['x_size'], self.FULLDISK_SIZES[0.01]['y_size'], self.AHI_FULLDISK_EXTENT) tmp_fh = self.make_fh('vis.01') tmp_fh.get_area_def(None) self.assertEqual(tmp_fh.area, good_area)
[docs] def test_bad_area(self): """Ensure an error is raised for an usupported area.""" tmp_fh = self.make_fh('ext.01') tmp_fh.areaname = 'scanning' with self.assertRaises(NotImplementedError): tmp_fh.get_area_def(None) with self.assertRaises(NotImplementedError): self.make_fh('ext.01', area='scanning')
[docs] class TestAHIGriddedFileCalibration(unittest.TestCase): """Test case for the file calibration types."""
[docs] def setUp(self): """Create a test file handler.""" m = mock.mock_open() with mock.patch('satpy.readers.ahi_l1b_gridded_bin.open', m, create=True): in_fname = 'test_file' fh = AHIGriddedFileHandler(in_fname, {'area': 'fld'}, filetype_info={'file_type': 'tir.01'}) self.fh = fh
[docs] @mock.patch('satpy.readers.ahi_l1b_gridded_bin.AHIGriddedFileHandler._get_luts') @mock.patch('satpy.readers.ahi_l1b_gridded_bin.os.path.exists') @mock.patch('satpy.readers.ahi_l1b_gridded_bin.np.loadtxt') def test_calibrate(self, np_loadtxt, os_exist, get_luts): """Test the calibration modes of AHI using the LUTs.""" load_return = np.squeeze(np.dstack([np.arange(0, 2048, 1), np.arange(0, 120, 0.05859375)])) np_loadtxt.return_value = load_return get_luts.return_value = True in_data = np.array([[100., 300., 500.], [800., 1500., 2040.]]) refl_out = np.array([[5.859375, 17.578125, 29.296875], [46.875, 87.890625, 119.53125]]) os_exist.return_value = False # Check that the LUT download is called if we don't have the LUTS self.fh.calibrate(in_data, 'reflectance') get_luts.assert_called() os_exist.return_value = True # Ensure results equal if no calibration applied out_data = self.fh.calibrate(in_data, 'counts') np.testing.assert_equal(in_data, out_data) # Now ensure results equal if LUT calibration applied out_data = self.fh.calibrate(in_data, 'reflectance') np.testing.assert_allclose(refl_out, out_data) # Check that exception is raised if bad calibration is passed with self.assertRaises(NotImplementedError): self.fh.calibrate(in_data, 'lasers') # Check that exception is raised if no file is present np_loadtxt.side_effect = FileNotFoundError with self.assertRaises(FileNotFoundError): self.fh.calibrate(in_data, 'reflectance')
[docs] class TestAHIGriddedFileHandler(unittest.TestCase): """Test case for the file reading."""
[docs] def new_unzip(fname): """Fake unzipping.""" if fname[-3:] == 'bz2': return fname[:-4]
[docs] @mock.patch('satpy.readers.ahi_l1b_gridded_bin.unzip_file', mock.MagicMock(side_effect=new_unzip)) def setUp(self): """Create a test file handler.""" m = mock.mock_open() with mock.patch('satpy.readers.ahi_l1b_gridded_bin.open', m, create=True): in_fname = 'test_file.bz2' fh = AHIGriddedFileHandler(in_fname, {'area': 'fld'}, filetype_info={'file_type': 'tir.01'}) # Check that the filename is altered for bz2 format files self.assertNotEqual(in_fname, fh.filename) self.fh = fh key = {'calibration': 'counts', 'name': 'vis.01'} info = {'units': 'unitless', 'standard_name': 'vis.01', 'wavelength': 10.8, 'resolution': 0.05} self.key = key self.info = info
[docs] @mock.patch('satpy.readers.ahi_l1b_gridded_bin.np.memmap') def test_dataread(self, memmap): """Check that a dask array is returned from the read function.""" test_arr = np.zeros((10, 10)) memmap.return_value = test_arr m = mock.mock_open() with mock.patch('satpy.readers.ahi_l1b_gridded_bin.open', m, create=True): res = self.fh._read_data(mock.MagicMock()) np.testing.assert_allclose(res, da.from_array(test_arr))
[docs] @mock.patch('satpy.readers.ahi_l1b_gridded_bin.AHIGriddedFileHandler._read_data') def test_get_dataset(self, mocked_read): """Check that a good dataset is returned on request.""" m = mock.mock_open() out_data = np.array([[100., 300., 500.], [800., 1500., 2040.]]) mocked_read.return_value = out_data with mock.patch('satpy.readers.ahi_l1b_gridded_bin.open', m, create=True): res = self.fh.get_dataset(self.key, self.info) mocked_read.assert_called() # Check output data is correct np.testing.assert_allclose(res.values, out_data) # Also check a couple of attributes self.assertEqual(res.attrs['name'], self.key['name']) self.assertEqual(res.attrs['wavelength'], self.info['wavelength'])
[docs] @mock.patch('os.path.exists', return_value=True) @mock.patch('os.remove') def test_destructor(self, exist_patch, remove_patch): """Check that file handler deletes files if needed.""" del self.fh remove_patch.assert_called()
[docs] class TestAHIGriddedLUTs(unittest.TestCase): """Test case for the downloading and preparing LUTs."""
[docs] def mocked_ftp_dl(fname): """Fake download of LUT tar file by creating a local tar.""" import os import tarfile import tempfile with tarfile.open(fname, "w:gz") as tar_handle: for namer in AHI_LUT_NAMES: tmpf = os.path.join(tempfile.tempdir, namer) with open(tmpf, 'w') as tmp_fid: tmp_fid.write("TEST\n") tar_handle.add(tmpf, arcname='count2tbb_v102/'+namer) os.remove(tmpf)
[docs] def setUp(self): """Create a test file handler.""" m = mock.mock_open() with mock.patch('satpy.readers.ahi_l1b_gridded_bin.open', m, create=True): in_fname = 'test_file' fh = AHIGriddedFileHandler(in_fname, {'area': 'fld'}, filetype_info={'file_type': 'tir.01'}) self.fh = fh key = {'calibration': 'counts', 'name': 'vis.01'} info = {'units': 'unitless', 'standard_name': 'vis.01', 'wavelength': 10.8, 'resolution': 0.05} self.key = key self.info = info
[docs] def tearDown(self): """Remove files and directories created by the tests.""" for lut_name in AHI_LUT_NAMES: tmp_filename = os.path.join(self.fh.lut_dir, lut_name) if os.path.isfile(tmp_filename): os.remove(tmp_filename) if os.path.isdir(self.fh.lut_dir): shutil.rmtree(self.fh.lut_dir)
[docs] @mock.patch('satpy.readers.ahi_l1b_gridded_bin.AHIGriddedFileHandler._download_luts', mock.MagicMock(side_effect=mocked_ftp_dl)) def test_get_luts(self): """Check that the function to download LUTs operates successfully.""" tempdir = tempfile.gettempdir() print(self.fh.lut_dir) self.fh._get_luts() self.assertFalse(os.path.exists(os.path.join(tempdir, 'count2tbb_v102/'))) for lut_name in AHI_LUT_NAMES: self.assertTrue(os.path.isfile(os.path.join(self.fh.lut_dir, lut_name)))
[docs] @mock.patch('urllib.request.urlopen') @mock.patch('shutil.copyfileobj') def test_download_luts(self, mock_dl, mock_shutil): """Test that the FTP library is called for downloading LUTS.""" m = mock.mock_open() with mock.patch('satpy.readers.ahi_l1b_gridded_bin.open', m, create=True): self.fh._download_luts('/test_file') mock_dl.assert_called() mock_shutil.assert_called()