test_mpi.py
# coding: utf-8
# This Python module is part of the PyRate software package.
#
# Copyright 2017 Geoscience Australia
#
# Licensed 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.
"""
This Python module contains tests for mpi operations in PyRate.
Tun this module as 'mpirun -n 4 pytest tests/test_mpi.py'
"""
import glob
import shutil
import numpy as np
import pytest
import os
import tempfile
import random
import string
import pyrate.core.orbital
import pyrate.core.shared
import tests.common
from pyrate import (
process, prepifg, merge, conv2tif)
from tests.common import (small_data_setup, reconstruct_mst,
reconstruct_linrate, SML_TEST_DEM_HDR_GAMMA, pre_prepare_ifgs)
from tests import common
from tests.test_covariance import legacy_maxvar
from pyrate.core import algorithm, ref_phs_est as rpe, mpiops, config as cf, covariance, refpixel
TRAVIS = True if 'TRAVIS' in os.environ else False
@pytest.fixture()
def tempdir():
"""
tempdir for tests
"""
def tmpdir():
return tempfile.mkdtemp()
return tmpdir
@pytest.fixture
def random_filename(tmpdir_factory):
def make_random_filename(ext=''):
dir = str(tmpdir_factory.mktemp('pyrate').realpath())
fname = ''.join(random.choice(string.ascii_lowercase) for _ in range(10))
return os.path.join(dir, fname + ext)
return make_random_filename
@pytest.fixture()
def get_config():
"""
Parameters
----------
conf_file: str
config file
Returns
-------
params: dict
dict of params
"""
def params(conf_file):
return cf.get_config_params(conf_file)
return params
# Make sure all MPI tests use this fixure
@pytest.fixture()
def mpisync(request):
mpiops.comm.barrier()
def fin():
mpiops.comm.barrier()
request.addfinalizer(fin)
return mpiops.comm
@pytest.fixture(params=[0, 1])
def roipac_or_gamma(request):
return request.param
@pytest.fixture(params=[1, 2])
def ref_est_method(request):
return request.param
@pytest.fixture(params=[1, 2, 5])
def row_splits(request):
return request.param
@pytest.fixture(params=[1, 2, 5])
def col_splits(request):
return request.param
@pytest.fixture(params=[1, 2, 5])
def modify_config(request, tempdir, get_config):
test_conf = common.TEST_CONF_ROIPAC
params_dict = get_config(test_conf)
params_dict[cf.IFG_LKSX] = request.param
params_dict[cf.IFG_LKSY] = request.param
params_dict[cf.OBS_DIR] = tempdir()
common.copytree(common.SML_TEST_GAMMA, params_dict[cf.OBS_DIR])
params_dict[cf.IFG_FILE_LIST] = os.path.join(params_dict[cf.OBS_DIR], 'ifms_17')
params_dict[cf.PARALLEL] = False
params_dict[cf.APS_CORRECTION] = 0
yield params_dict
# clean up
shutil.rmtree(params_dict[cf.OBS_DIR])
@pytest.fixture(params=range(1, 6))
def get_lks(request):
return request.param
@pytest.fixture(params=range(1, 3))
def get_crop(request):
return request.param
def test_vcm_legacy_vs_mpi(mpisync, tempdir, get_config):
from tests.common import SML_TEST_DIR, TEST_CONF_ROIPAC
params_dict = get_config(TEST_CONF_ROIPAC)
LEGACY_VCM_DIR = os.path.join(SML_TEST_DIR, 'vcm')
legacy_vcm = np.genfromtxt(os.path.join(LEGACY_VCM_DIR, 'vcmt.csv'), delimiter=',')
if mpiops.rank == 0:
outdir = tempdir()
else:
outdir = None
outdir = mpiops.comm.bcast(outdir, root=0)
params_dict[cf.OUT_DIR] = outdir
params_dict[cf.PARALLEL] = False
xlks, ylks, crop = cf.transform_params(params_dict)
base_unw_paths = cf.original_ifg_paths(params_dict[cf.IFG_FILE_LIST], params_dict[cf.OBS_DIR])
# dest_paths are tifs that have been geotif converted and multilooked
dest_paths = cf.get_dest_paths(base_unw_paths, crop, params_dict, xlks)
# run prepifg, create the dest_paths files
if mpiops.rank == 0:
conv2tif.main(params_dict)
prepifg.main(params_dict)
mpiops.comm.barrier()
tiles = pyrate.core.shared.get_tiles(dest_paths[0], rows=1, cols=1)
preread_ifgs = process._create_ifg_dict(dest_paths, params=params_dict, tiles=tiles)
refpx, refpy = process._ref_pixel_calc(dest_paths, params_dict)
process._orb_fit_calc(dest_paths, params_dict)
process._ref_phase_estimation(dest_paths, params_dict, refpx, refpy)
maxvar, vcmt = process._maxvar_vcm_calc(dest_paths, params_dict, preread_ifgs)
np.testing.assert_array_almost_equal(maxvar, legacy_maxvar, decimal=4)
np.testing.assert_array_almost_equal(legacy_vcm, vcmt, decimal=3)
if mpiops.rank == 0:
shutil.rmtree(outdir)
common.remove_tifs(params_dict[cf.OBS_DIR])
@pytest.fixture(params=[1, 2, 5])
def orbfit_lks(request):
return request.param
@pytest.fixture(params=[cf.INDEPENDENT_METHOD, cf.NETWORK_METHOD])
def orbfit_method(request):
return request.param
@pytest.fixture(params=[cf.PLANAR, cf.QUADRATIC, cf.PART_CUBIC])
def orbfit_degrees(request):
return request.param
def _tifs_same(dir1, dir2, tif):
linrate_tif_s = os.path.join(dir1, tif)
linrate_tif_m = os.path.join(dir2, tif)
common.assert_ifg_phase_equal(linrate_tif_m, linrate_tif_s)
def test_prepifg_mpi(mpisync, get_config, tempdir,
roipac_or_gamma, get_lks, get_crop):
from tests.common import TEST_CONF_ROIPAC, TEST_CONF_GAMMA
from os.path import join, basename
if roipac_or_gamma == 1:
params = get_config(TEST_CONF_GAMMA)
else:
params = get_config(TEST_CONF_ROIPAC)
outdir = mpiops.run_once(tempdir)
params[cf.OUT_DIR] = outdir
params[cf.PARALLEL] = False
params[cf.IFG_LKSX], params[cf.IFG_LKSY] = get_lks, get_lks
params[cf.IFG_CROP_OPT] = get_crop
if roipac_or_gamma == 1:
params[cf.IFG_FILE_LIST] = join(common.SML_TEST_GAMMA, 'ifms_17')
params[cf.OBS_DIR] = common.SML_TEST_GAMMA
params[cf.DEM_FILE] = common.SML_TEST_DEM_GAMMA
params[cf.DEM_HEADER_FILE] = common.SML_TEST_DEM_HDR_GAMMA
conv2tif.main(params)
prepifg.main(params)
common.remove_tifs(params[cf.OBS_DIR])
if mpiops.rank == 0:
if roipac_or_gamma == 1:
params_s = get_config(TEST_CONF_GAMMA)
else:
params_s = get_config(TEST_CONF_ROIPAC)
params_s[cf.OUT_DIR] = tempdir()
params_s[cf.PARALLEL] = True
params_s[cf.IFG_LKSX], params_s[cf.IFG_LKSY] = get_lks, get_lks
params_s[cf.IFG_CROP_OPT] = get_crop
conv2tif.main(params)
if roipac_or_gamma == 1:
base_unw_paths = glob.glob(join(common.SML_TEST_GAMMA,
"*_utm.unw"))
prepifg.main(params)
else:
base_unw_paths = glob.glob(join(common.SML_TEST_OBS, "*.unw"))
prepifg.main(params_s)
mpi_tifs = glob.glob(join(outdir, "*.tif"))
serial_tifs = glob.glob(join(params[cf.OUT_DIR], "*.tif"))
mpi_tifs.sort()
serial_tifs.sort()
# 17 geotifs, and 17 mlooked tifs
assert len(mpi_tifs) == len(serial_tifs)
for m_f, s_f in zip(mpi_tifs, serial_tifs):
assert basename(m_f) == basename(s_f)
shutil.rmtree(outdir)
shutil.rmtree(params_s[cf.OUT_DIR])
common.remove_tifs(params[cf.OBS_DIR])
