blob: 3731d849d25850a5b8abd757218bc3db25d49eed [file] [log] [blame]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import unittest
import numpy
import numpy.ma as ma
import os
import netCDF4
import datetime
from data_source.local import *
class test_load_file(unittest.TestCase):
def setUp(self):
#To create the temporary netCDF file
self.file_path = os.getcwd() + '/temporaryNetcdf.nc'
netCDF_file = netCDF4.Dataset(self.file_path, 'w', format='NETCDF4')
#To create dimensions
netCDF_file.createDimension('lat_dim', 5)
netCDF_file.createDimension('lon_dim', 5)
netCDF_file.createDimension('time_dim', 3)
netCDF_file.createDimension('level_dim', 2)
#To create variables
latitudes = netCDF_file.createVariable('latitude', 'd', ('lat_dim',))
longitudes = netCDF_file.createVariable('longitude', 'd', ('lon_dim',))
times = netCDF_file.createVariable('time', 'd', ('time_dim',))
levels = netCDF_file.createVariable('level', 'd', ('level_dim',))
values = netCDF_file.createVariable('value', 'd', ('level_dim', 'time_dim', 'lat_dim', 'lon_dim'))
#To latitudes and longitudes for five values
self.latitudes = range(0,5)
self.longitudes = range(200,205)
#Three months of data
self.times = range(3)
#Two levels
self.levels = [100, 200]
#Create 150 values
self.values = numpy.array([i for i in range(150)])
#Reshape values to 4D array (level, time, lats, lons)
self.values = self.values.reshape(len(self.levels), len(self.times),len(self.latitudes),len(self.longitudes))
#Ingest values to netCDF file
latitudes[:] = self.latitudes
longitudes[:] = self.longitudes
times[:] = self.times
levels[:] = self.levels
values[:] = self.values
#Assigne time info to time variable
netCDF_file.variables['time'].units = 'months since 2001-01-01 00:00:00'
netCDF_file.close()
#Read netCDF file
self.netCDF_file = netCDF4.Dataset(self.file_path, 'r')
self.user_value_variable_name = 'value'
self.variable_name_list = ['latitude', 'longitude', 'time', 'level', 'value']
self.possible_value_name = ['latitude', 'longitude', 'time', 'level']
def tearDown(self):
'''To remove the temporary netCDF file'''
os.remove(self.file_path)
def test_function_load_file_lats(self):
'''To test load_file function for latitudes'''
self.assertItemsEqual(load_file(self.file_path, None).lats, self.latitudes)
def test_function_load_file_lons(self):
'''To test load_file function for longitudes'''
self.assertItemsEqual(load_file(self.file_path, None).lons, self.longitudes)
def test_function_load_file_times(self):
'''To test load_file function for times'''
newTimes = datetime(2001,01,01), datetime(2001,02,01), datetime(2001,03,01)
self.assertItemsEqual(load_file(self.file_path, None).times, newTimes)
def test_function_load_file_values(self):
'''To test load_file function for values'''
new_values = self.values[0,:,:,:]
self.assertTrue(numpy.allclose(load_file(self.file_path, None).values, new_values))
if __name__ == '__main__':
unittest.main()