How to use load_file method in localstack

Best Python code snippet using localstack_python

test_im_calculation.py

Source:test_im_calculation.py

1import argparse2import filecmp3import os4import pickle5import pytest6import numpy as np7from qcore.constants import Components8import IM_calculation.IM.im_calculation as calculate_ims9from qcore import utils, constants10from IM_calculation.test.test_common_set_up import INPUT, OUTPUT, compare_dicts, set_up11# This is a hack, to allow loading of the test pickle objects12import sys13import IM_calculation.IM as IM14sys.modules["IM"] = IM15PARSER = argparse.ArgumentParser()16BSC_PERIOD = [0.05, 0.1, 5.0, 10.0]17TEST_IMS = ["PGA", "PGV", "Ds575", "pSA"]18FAKE_DIR = (19    "fake_dir"20)  # should be created in set_up module and remove in tear_down module21utils.setup_dir("fake_dir")22@pytest.mark.parametrize(23    "test_period, test_extended, expected_period",24    [25        (BSC_PERIOD, False, np.array(BSC_PERIOD)),26        (BSC_PERIOD, True, np.unique(np.append(BSC_PERIOD, constants.EXT_PERIOD))),27    ],28)29def test_validate_period(test_period, test_extended, expected_period):30    assert all(31        np.equal(32            calculate_ims.validate_period(test_period, test_extended), expected_period33        )34    )35@pytest.mark.parametrize("test_path, test_file_type", [("asdf", "b"), (FAKE_DIR, "b")])36def test_validate_input_path_fail(test_path, test_file_type):37    with pytest.raises(SystemExit):38        calculate_ims.validate_input_path(PARSER, test_path, test_file_type)39def convert_str_comps_to_enum(expected_result):40    for station in expected_result.keys():41        for im in expected_result[station].keys():42            if im == "pSA":43                for comp in list(expected_result[station][im][1]):44                    expected_result[station][im][1][45                        Components.from_str(comp)46                    ] = expected_result[station][im][1][comp]47                    del expected_result[station][im][1][comp]48            else:49                for comp in list(expected_result[station][im]):50                    expected_result[station][im][51                        Components.from_str(comp)52                    ] = expected_result[station][im][comp]53                    del expected_result[station][im][comp]54class TestPickleTesting:55    def test_convert_str_comp(self, set_up):56        function = "convert_str_comp"57        for root_path in set_up:58            with open(59                os.path.join(root_path, INPUT, function + "_comp.P"), "rb"60            ) as load_file:61                comp = pickle.load(load_file)62            int_comp, str_comp = Components.get_comps_to_calc_and_store(comp)63            with open(64                os.path.join(root_path, OUTPUT, function + "_str_comp_for_int.P"), "rb"65            ) as load_file:66                expected_int_comp = pickle.load(load_file)67            with open(68                os.path.join(root_path, OUTPUT, function + "_str_comp.P"), "rb"69            ) as load_file:70                expected_str_comp = pickle.load(load_file)71            assert [x.str_value for x in int_comp] == expected_int_comp72            assert [x.str_value for x in str_comp] == expected_str_comp73    def test_array_to_dict(self, set_up):74        function = "array_to_dict"75        for root_path in set_up:76            with open(77                os.path.join(root_path, INPUT, function + "_value.P"), "rb"78            ) as load_file:79                value = pickle.load(load_file)80            with open(81                os.path.join(root_path, INPUT, function + "_comp.P"), "rb"82            ) as load_file:83                arg_comps = pickle.load(load_file)84            with open(85                os.path.join(root_path, INPUT, function + "_str_comp.P"), "rb"86            ) as load_file:87                str_comps = pickle.load(load_file)88            with open(89                os.path.join(root_path, INPUT, function + "_im.P"), "rb"90            ) as load_file:91                im = pickle.load(load_file)92            str_comps = [Components.from_str(x) for x in str_comps]93            arg_comps = [Components.from_str(x) for x in arg_comps]94            actual_value_dict = calculate_ims.array_to_dict(value, str_comps, im, arg_comps)95            with open(96                os.path.join(root_path, OUTPUT, function + "_value_dict.P"), "rb"97            ) as load_file:98                expected_value_dict = pickle.load(load_file)99            assert actual_value_dict == expected_value_dict100    def test_compute_measure_single(self, set_up):101        function = "compute_measure_single"102        for root_path in set_up:103            with open(104                os.path.join(root_path, INPUT, function + "_value_tuple.P"), "rb"105            ) as load_file:106                value_tuple = pickle.load(load_file)107            waveform, ims, comps, periods, str_comps = value_tuple108            im_options = {"pSA": periods}109            comps = [Components.from_str(x) for x in comps]110            str_comps = [Components.from_str(x) for x in str_comps]111            actual_result = calculate_ims.compute_measure_single(112                waveform, ims, comps, im_options, str_comps, (0,0)113            )114            with open(115                os.path.join(root_path, OUTPUT, function + "_result.P"), "rb"116            ) as load_file:117                expected_result = pickle.load(load_file)118                convert_str_comps_to_enum(expected_result)119                actual_expected_result = self.convert_to_results_dict(periods, expected_result)120            compare_dicts(actual_result, actual_expected_result)121    def test_get_bbseis(self, set_up):122        function = "get_bbseis"123        for root_path in set_up:124            with open(125                os.path.join(root_path, INPUT, function + "_selected_stations.P"), "rb"126            ) as load_file:127                stations = pickle.load(load_file)128            actual_converted_stations = calculate_ims.get_bbseis(129                os.path.join(root_path, INPUT, "BB.bin"), "binary", stations130            )[1]131            with open(132                os.path.join(root_path, OUTPUT, function + "_station_names.P"), "rb"133            ) as load_file:134                expected_converted_stations = pickle.load(load_file)135            assert actual_converted_stations == expected_converted_stations136    def test_compute_measures_multiprocess(self, set_up):137        function = "compute_measures_multiprocess"138        for root_path in set_up:139            input_path = os.path.join(root_path, INPUT, "BB.bin")140            with open(141                os.path.join(root_path, INPUT, function + "_file_type.P"), "rb"142            ) as load_file:143                file_type = pickle.load(load_file)144            with open(145                os.path.join(root_path, INPUT, function + "_wave_type.P"), "rb"146            ) as load_file:147                wave_type = pickle.load(load_file)148            with open(149                os.path.join(root_path, INPUT, function + "_ims.P"), "rb"150            ) as load_file:151                ims = pickle.load(load_file)152            with open(153                os.path.join(root_path, INPUT, function + "_comp.P"), "rb"154            ) as load_file:155                comp = pickle.load(load_file)156            with open(157                os.path.join(root_path, INPUT, function + "_period.P"), "rb"158            ) as load_file:159                period = pickle.load(load_file)160            with open(161                os.path.join(root_path, INPUT, function + "_identifier.P"), "rb"162            ) as load_file:163                identifier = pickle.load(load_file)164            with open(165                os.path.join(root_path, INPUT, function + "_rupture.P"), "rb"166            ) as load_file:167                rupture = pickle.load(load_file)168            with open(169                os.path.join(root_path, INPUT, function + "_run_type.P"), "rb"170            ) as load_file:171                run_type = pickle.load(load_file)172            with open(173                os.path.join(root_path, INPUT, function + "_version.P"), "rb"174            ) as load_file:175                version = pickle.load(load_file)176            with open(177                os.path.join(root_path, INPUT, function + "_process.P"), "rb"178            ) as load_file:179                process = pickle.load(load_file)180            with open(181                os.path.join(root_path, INPUT, function + "_simple_output.P"), "rb"182            ) as load_file:183                simple_output = pickle.load(load_file)184            station_names = ["099A"]185            output = root_path186            os.makedirs(os.path.join(output, "stations"), exist_ok=True)187            calculate_ims.compute_measures_multiprocess(188                input_path,189                file_type,190                wave_type,191                station_names,192                ims,193                comp,194                {"pSA": period},195                output,196                identifier,197                rupture,198                run_type,199                version,200                process,201                simple_output,202            )203    def test_get_result_filepath(self, set_up):204        function = "get_result_filepath"205        for root_path in set_up:206            with open(207                os.path.join(root_path, INPUT, function + "_output_folder.P"), "rb"208            ) as load_file:209                output_folder = pickle.load(load_file)210            with open(211                os.path.join(root_path, INPUT, function + "_arg_identifier.P"), "rb"212            ) as load_file:213                arg_identifier = pickle.load(load_file)214            with open(215                os.path.join(root_path, INPUT, function + "_suffix.P"), "rb"216            ) as load_file:217                suffix = pickle.load(load_file)218            actual_ret_val = calculate_ims.get_result_filepath(219                output_folder, arg_identifier, suffix220            )221            with open(222                os.path.join(root_path, OUTPUT, function + "_ret_val.P"), "rb"223            ) as load_file:224                expected_ret_val = pickle.load(load_file)225            assert actual_ret_val == expected_ret_val226    def test_write_result(self, set_up):227        function = "write_result"228        for root_path in set_up:229            with open(230                os.path.join(root_path, INPUT, function + "_period.P"), "rb"231            ) as load_file:232                period = pickle.load(load_file)233            with open(234                os.path.join(root_path, INPUT, function + "_result_dict.P"), "rb"235            ) as load_file:236                temp_result_dict = pickle.load(load_file)237                convert_str_comps_to_enum(temp_result_dict)238                result_dict = self.convert_to_results_dict(period, temp_result_dict, keep_ps=True)239            with open(240                os.path.join(root_path, INPUT, function + "_identifier.P"), "rb"241            ) as load_file:242                identifier = pickle.load(load_file)243            with open(244                os.path.join(root_path, INPUT, function + "_simple_output.P"), "rb"245            ) as load_file:246                simple_output = pickle.load(load_file)247            output_folder = root_path248            os.makedirs(249                os.path.join(output_folder, calculate_ims.OUTPUT_SUBFOLDER),250                exist_ok=True,251            )252            calculate_ims.write_result(result_dict, output_folder, identifier, simple_output)253            expected_output_path = calculate_ims.get_result_filepath(254                output_folder, identifier, ".csv"255            )256            actual_output_path = os.path.join(257                root_path, OUTPUT, function + "_outfile.csv"258            )259            expected_output = np.loadtxt(expected_output_path, delimiter=',',usecols=range(2,24), skiprows=1)260            actual_output = np.loadtxt(actual_output_path, delimiter=',',usecols=range(2,24), skiprows=1)261            assert np.isclose(expected_output, actual_output).all()262    def convert_to_results_dict(self, period, temp_result_dict, keep_ps=False):263        result_dict = {}264        for station in sorted(temp_result_dict):265            temp_result_dict[station]["pSA"] = temp_result_dict[station]["pSA"][1]266            for im in sorted(temp_result_dict[station]):267                for comp in temp_result_dict[station][im]:268                    if (station, comp.str_value) not in result_dict:269                        result_dict[(station, comp.str_value)] = {}270                    if im in calculate_ims.MULTI_VALUE_IMS:271                        for i, val in enumerate(period):272                            if keep_ps:273                                result_dict[(station, comp.str_value)][f"{im}_{str(val).replace('.', 'p')}"] = \274                                temp_result_dict[station][im][comp][i]275                            else:276                                result_dict[(station, comp.str_value)][f"{im}_{str(val)}"] = \277                                temp_result_dict[station][im][comp][i]278                    else:279                        result_dict[(station, comp.str_value)][im] = temp_result_dict[station][im][comp]280        return result_dict281    def test_generate_metadata(self, set_up):282        function = "generate_metadata"283        for root_path in set_up:284            with open(285                os.path.join(root_path, INPUT, function + "_identifier.P"), "rb"286            ) as load_file:287                identifier = pickle.load(load_file)288            with open(289                os.path.join(root_path, INPUT, function + "_rupture.P"), "rb"290            ) as load_file:291                rupture = pickle.load(load_file)292            with open(293                os.path.join(root_path, INPUT, function + "_run_type.P"), "rb"294            ) as load_file:295                run_type = pickle.load(load_file)296            with open(297                os.path.join(root_path, INPUT, function + "_version.P"), "rb"298            ) as load_file:299                version = pickle.load(load_file)300            # Save to the realisations folder that will be deleted after the run has finished301            output_folder = root_path302            calculate_ims.generate_metadata(303                output_folder, identifier, rupture, run_type, version304            )305            actual_output_path = calculate_ims.get_result_filepath(306                output_folder, identifier, "_imcalc.info"307            )308            expected_output_path = os.path.join(309                root_path, OUTPUT, function + "_outfile.info"310            )311            filecmp.cmp(actual_output_path, expected_output_path)312    def test_validate_input_path(self, set_up):313        function = "validate_input_path"314        for root_path in set_up:315            arg_input = os.path.join(root_path, INPUT, "BB.bin")316            with open(317                os.path.join(root_path, INPUT, function + "_arg_file_type.P"), "rb"318            ) as load_file:319                arg_file_type = pickle.load(load_file)320            calculate_ims.validate_input_path(PARSER, arg_input, arg_file_type)321            # Function does not return anything, only raises errors through the parser322    def test_validate_period(self, set_up):323        function = "validate_period"324        for root_path in set_up:325            with open(326                os.path.join(root_path, INPUT, function + "_arg_period.P"), "rb"327            ) as load_file:328                arg_period = pickle.load(load_file)329            with open(330                os.path.join(root_path, INPUT, function + "_arg_extended_period.P"),331                "rb",332            ) as load_file:333                arg_extended_period = pickle.load(load_file)334            actual_period = calculate_ims.validate_period(335                arg_period, arg_extended_period336            )337            with open(338                os.path.join(root_path, OUTPUT, function + "_period.P"), "rb"339            ) as load_file:340                expected_period = pickle.load(load_file)341            assert (actual_period == expected_period).all()342    def test_get_steps(self, set_up):343        function = "get_steps"344        for root_path in set_up:345            input_path = os.path.join(root_path, INPUT, "BB.bin")346            with open(347                os.path.join(root_path, INPUT, function + "_nps.P"), "rb"348            ) as load_file:349                nps = pickle.load(load_file)350            with open(351                os.path.join(root_path, INPUT, function + "_total_stations.P"), "rb"352            ) as load_file:353                total_stations = pickle.load(load_file)354            actual_steps = calculate_ims.get_steps(input_path, nps, total_stations)355            with open(356                os.path.join(root_path, OUTPUT, function + "_steps.P"), "rb"357            ) as load_file:358                expected_steps = pickle.load(load_file)...

test_intensity_measures.py

Source:test_intensity_measures.py

1import os2import pickle3import numpy as np4import pytest5from IM_calculation.IM import intensity_measures6from IM_calculation.test.test_common_set_up import INPUT, OUTPUT, set_up7def get_common_spectral_vals(root_path, function_name):8    with open(9        os.path.join(root_path, INPUT, function_name + "_acceleration.P"), "rb"10    ) as load_file:11        acc = pickle.load(load_file)12    with open(13        os.path.join(root_path, INPUT, function_name + "_period.P"), "rb"14    ) as load_file:15        period = pickle.load(load_file)16    with open(17        os.path.join(root_path, INPUT, function_name + "_NT.P"), "rb"18    ) as load_file:19        NT = pickle.load(load_file)20    with open(21        os.path.join(root_path, INPUT, function_name + "_DT.P"), "rb"22    ) as load_file:23        DT = pickle.load(load_file)24    return acc, period, NT, DT25def get_common_vals(root_path, function_name):26    with open(27        os.path.join(root_path, INPUT, function_name + "_acceleration.P"), "rb"28    ) as load_file:29        acc = pickle.load(load_file)30    with open(31        os.path.join(root_path, INPUT, function_name + "_times.P"), "rb"32    ) as load_file:33        times = pickle.load(load_file)34    return acc, times35def get_common_ds_vals(root_path, function_name):36    with open(37        os.path.join(root_path, INPUT, function_name + "_dt.P"), "rb"38    ) as load_file:39        dt = pickle.load(load_file)40    with open(41        os.path.join(root_path, INPUT, function_name + "_percLow.P"), "rb"42    ) as load_file:43        perc_low = pickle.load(load_file)44    with open(45        os.path.join(root_path, INPUT, function_name + "_percHigh.P"), "rb"46    ) as load_file:47        perc_high = pickle.load(load_file)48    return dt, perc_low, perc_high49def test_get_max_nd(set_up):50    function = "get_max_nd"51    for root_path in set_up:52        with open(53            os.path.join(root_path, INPUT, function + "_data.P"), "rb"54        ) as load_file:55            data = pickle.load(load_file)56        test_output = intensity_measures.get_max_nd(data)57        with open(58            os.path.join(root_path, OUTPUT, function + "_ret_val.P"), "rb"59        ) as load_file:60            bench_output = pickle.load(load_file)61        assert np.isclose(test_output, bench_output).all()62def test_get_spectral_acceleration(set_up):63    function = "get_spectral_acceleration"64    for root_path in set_up:65        acc, period, NT, DT = get_common_spectral_vals(root_path, function)66        test_output = intensity_measures.get_spectral_acceleration(67            acc, period, NT, DT, intensity_measures.calculate_Nstep(DT, NT)68        )69        with open(70            os.path.join(root_path, OUTPUT, function + "_ret_val.P"), "rb"71        ) as load_file:72            bench_output = pickle.load(load_file)73        assert np.isclose(test_output, bench_output).all()74def test_get_spectral_acceleration_nd(set_up):75    function = "get_spectral_acceleration_nd"76    for root_path in set_up:77        acc, period, NT, DT = get_common_spectral_vals(root_path, function)78        test_output = intensity_measures.get_spectral_acceleration_nd(79            acc, period, NT, DT80        )81        with open(82            os.path.join(root_path, OUTPUT, function + "_values.P"), "rb"83        ) as load_file:84            bench_output = pickle.load(load_file)85        assert np.isclose(test_output, bench_output).all()86def test_get_cumulative_abs_velocity_nd(set_up):87    function = "get_cumulative_abs_velocity_nd"88    for root_path in set_up:89        acc, times = get_common_vals(root_path, function)90        test_output = intensity_measures.get_cumulative_abs_velocity_nd(acc, times)91        with open(92            os.path.join(root_path, OUTPUT, function + "_ret_val.P"), "rb"93        ) as load_file:94            bench_output = pickle.load(load_file)95        assert np.isclose(test_output, bench_output).all()96def test_get_arias_intensity_nd(set_up):97    function = "get_arias_intensity_nd"98    for root_path in set_up:99        acc, times = get_common_vals(root_path, function)100        test_output = intensity_measures.get_arias_intensity_nd(acc, times)101        with open(102            os.path.join(root_path, OUTPUT, function + "_ret_val.P"), "rb"103        ) as load_file:104            bench_output = pickle.load(load_file)105        assert np.isclose(test_output, bench_output).all()106def test_calculate_MMI_nd(set_up):107    function = "calculate_MMI_nd"108    for root_path in set_up:109        with open(110            os.path.join(root_path, INPUT, function + "_velocities.P"), "rb"111        ) as load_file:112            vel = pickle.load(load_file)113        test_output = intensity_measures.calculate_MMI_nd(vel)114        with open(115            os.path.join(root_path, OUTPUT, function + "_ret_val.P"), "rb"116        ) as load_file:117            bench_output = pickle.load(load_file)118        assert np.isclose(test_output, bench_output).all()119def test_getDs(set_up):120    function = "getDs"121    for root_path in set_up:122        dt, perc_low, perc_high = get_common_ds_vals(root_path, function)123        with open(124            os.path.join(root_path, INPUT, function + "_fx.P"), "rb"125        ) as load_file:126            fx = pickle.load(load_file)127        test_output = intensity_measures.getDs(dt, fx, perc_low, perc_high)128        with open(129            os.path.join(root_path, OUTPUT, function + "_Ds.P"), "rb"130        ) as load_file:131            bench_output = pickle.load(load_file)132        assert np.isclose(test_output, bench_output).all()133def test_getDs_nd(set_up):134    function = "getDs_nd"135    for root_path in set_up:136        dt, perc_low, perc_high = get_common_ds_vals(root_path, function)137        with open(138            os.path.join(root_path, INPUT, function + "_accelerations.P"), "rb"139        ) as load_file:140            acc = pickle.load(load_file)141        test_output = intensity_measures.getDs_nd(acc, dt, perc_low, perc_high)142        with open(143            os.path.join(root_path, OUTPUT, function + "_values.P"), "rb"144        ) as load_file:145            bench_output = pickle.load(load_file)146        assert np.isclose(test_output, bench_output).all()147def test_get_geom(set_up):148    function = "get_geom"149    for root_path in set_up:150        with open(151            os.path.join(root_path, INPUT, function + "_d1.P"), "rb"152        ) as load_file:153            d1 = pickle.load(load_file)154        with open(155            os.path.join(root_path, INPUT, function + "_d2.P"), "rb"156        ) as load_file:157            d2 = pickle.load(load_file)158        test_output = intensity_measures.get_geom(d1, d2)159        with open(160            os.path.join(root_path, OUTPUT, function + "_ret_val.P"), "rb"161        ) as load_file:162            bench_output = pickle.load(load_file)163        assert np.isclose(test_output, bench_output).all()164@pytest.mark.parametrize(165    "test_d1, test_d2, expected_geom", [(0, 0, 0), (1, 5, 2.236067)]166)167def test_get_geom_params(test_d1, test_d2, expected_geom):168    assert np.isclose(169        intensity_measures.get_geom(test_d1, test_d2), expected_geom...

data_loader.py

Source:data_loader.py

1import csv2def load_file(file_name,data,labels):3    with open(file_name, 'r') as ppd:4        for line in ppd:5            attr = line.split(',')6            data.append(list(map(float, attr[0:31])))7            labels.append(attr[31][0:-1])8def data_loader():9    data = []10    labels = []11    load_file("../old/master_data/Alice/g_m.csv",data,labels)12    load_file("../old/master_data/Alice/n_s.csv",data,labels)13    load_file("../old/master_data/Alice/t_y.csv",data,labels)14    load_file("../old/master_data/Anisha/a_f.csv",data,labels)15    load_file("../old/master_data/Anisha/g_m.csv",data,labels)16    load_file("../old/master_data/Anisha/n_s.csv",data,labels)17    load_file("../old/master_data/Anisha/t_y.csv",data,labels)18    load_file("../old/master_data/Asutosh/a_f.csv",data,labels)19    load_file("../old/master_data/Asutosh/g_m.csv",data,labels)20    load_file("../old/master_data/Asutosh/n_s.csv",data,labels)21    load_file("../old/master_data/Asutosh/t_y.csv",data,labels)22    load_file("../old/master_data/Rishav/a_f.csv",data,labels)23    load_file("../old/master_data/Rishav/g_m.csv",data,labels)24    load_file("../old/master_data/Rishav/n_s.csv",data,labels)25    load_file("../old/master_data/Rishav/t_y.csv",data,labels)26    load_file("../old/master_data/Sai/a_f.csv",data,labels)27    load_file("../old/master_data/Sai/g_m.csv",data,labels)28    load_file("../old/master_data/Sai/n_s.csv",data,labels)29    load_file("../old/master_data/Sai/t_y.csv",data,labels)30    load_file("../old/master_data/Sandy/a_f.csv",data,labels)31    load_file("../old/master_data/Sandy/g_m.csv",data,labels)32    load_file("../old/master_data/Sandy/n_s.csv",data,labels)33    load_file("../old/master_data/Sandy/t_y.csv",data,labels)34    load_file("../old/master_data/Sohini/a_f.csv",data,labels)35    load_file("../old/master_data/Sohini/g_m.csv",data,labels)36    load_file("../old/master_data/Sohini/n_s.csv",data,labels)37    load_file("../old/master_data/Sohini/t_y.csv",data,labels)38    load_file("../old/master_data/zairza.csv",data,labels)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.