import pytest import os from os import path import netCDF4 from typing import List, Set, Dict, Tuple, Optional from tests.TestUtils import TestUtils from netCDF4 import Dataset from SDToolBox import main as main from SDToolBox.input_data import InputData from SDToolBox.extract_data import ExtractData import SDToolBox.output_messages as om class Test_get_era5: @pytest.mark.systemtest def test_given_list_of_coordinates_then_subset_is_extracted(self): # 1. Given # When using local data you can just replace the comment in these lines dir_test_data = TestUtils.get_local_test_data_dir('era5_test_data') # dir_test_data = 'P:\\metocean-data\\open\\ERA5\\data\\Global' input_data = InputData() input_data.input_variables = ['swh'] input_data.input_years = [1981, 1982] input_data.input_coordinates = \ [(4.2, 2.4), (42, 2.4), (42, 24), (4.2, 24)] # 2. When output_data = ExtractData.get_era_5( dir_test_data, input_data) # 3. Then assert output_data is not None data_dict = output_data.data_dict assert data_dict is not None assert data_dict[output_data.var_time_key] is not None assert data_dict[output_data.var_lon_key] is not None assert data_dict[output_data.var_lat_key] is not None assert data_dict[output_data.var_val_key] is not None assert data_dict[output_data.var_val_key]['swh'] is not None @pytest.mark.systemtest @pytest.mark.parametrize( "input_variables", [(['swh']), (['u10']), (['msl_p'])], ids=['Waves', 'Wind', 'SLP']) def test_when_extract_single_point_from_era5_dir_then_returns_output( self, input_variables: List[str]): # 1. Given # When using local data you can just replace the comment in these lines dir_test_data = TestUtils.get_local_test_data_dir('era5_test_data') # dir_test_data = 'P:\\metocean-data\\open\\ERA5\\data\\Global' input_data = InputData() input_data.input_variables = input_variables input_data.input_coordinates = [(4.2, 2.4), ] input_data.input_years = [1981, 1982] # 2. When try: output_data = ExtractData.get_era_5( dir_test_data, input_data) except Exception as e_info: pytest.fail('Exception {} thrown.'.format(str(e_info))) # 3. Then assert output_data is not None data_dict = output_data.data_dict assert data_dict is not None assert data_dict[output_data.var_time_key] is not None assert data_dict[output_data.var_lon_key] is not None assert data_dict[output_data.var_lat_key] is not None assert data_dict[output_data.var_val_key] is not None assert data_dict[output_data.var_val_key]['swh'] is not None class Test_get_earth: @pytest.mark.systemtest @pytest.mark.parametrize( "input_variables, input_scenarios", [(['var151'], ['RCP45'])], ids=['RCP45 simple']) def test_when_extract_single_point_from_earth_dir_then_returns_output( self, input_variables: List[str], input_scenarios: List[str]): # 1. Given # When using local data you can just replace the comment in these lines dir_test_data = TestUtils.get_local_test_data_dir('earth_test_data') # dir_test_data = 'P:\\metocean-data\\open\\ERA5\\data\\Global' input_data = InputData() input_data.input_variables = input_variables input_data.input_coordinates = [(4.2, 2.4), ] input_data.input_years = [1981, 1982] input_data.input_scenarios = input_scenarios # 2. When try: output_data = ExtractData.get_earth( dir_test_data, input_data) except Exception as e_info: pytest.fail('Exception {} thrown.'.format(str(e_info))) # 3. Then assert output_data is not None data_dict = output_data.data_dict assert data_dict is not None assert data_dict[output_data.var_time_key] is not None assert data_dict[output_data.var_lon_key] is not None assert data_dict[output_data.var_lat_key] is not None assert data_dict[output_data.var_val_key] is not None