TEST : Add tests for resampling

python_src/tests/test_resampling.py

+import pytest
+import shutil
+import os
+from copy import deepcopy
+from enum import Enum
+
+try:
+    import gdal
+except ImportError:
+    import osgeo.gdal as gdal
+
+
+from diapotb.lib.core.Utils import MetadataKeyInGeom
+from diapotb.lib.internal_processing.resampling import retrieve_frequencies, main_resampling
+from diapotb.lib.internal_processing.core.resampling_dimensions import ResamplingDimension
+from diapotb.lib.internal_processing.core.processing_dimensions import Dimension
+import diapotb.lib.internal_processing.core.utils as utils
+
+@pytest.fixture(scope="function")
+def create_dict_kwl(same_frequencies, dimension):
+    """Build two kwl (reference and secondary) following a boolean : same_frequencies"""
+    # Select the metadata key to set elements following dimension
+    metadata_key = str(MetadataKeyInGeom.FREQ_SAMPLING)
+    # dummy value
+    metadata_value = 4242
+    if dimension == Dimension.AZIMUT:
+        metadata_key = str(MetadataKeyInGeom.LINE_INTERVAL)
+        metadata_value = 1/metadata_value
+
+    # Set kwl_ref and kwl_sec
+    kwl_ref = {metadata_key: metadata_value}
+
+    if same_frequencies:
+        # Same value
+        kwl_sec = {metadata_key: metadata_value}
+    else:
+        # Different value
+        delta_value = 25
+        if dimension == Dimension.AZIMUT:
+            delta_value = 1/delta_value
+
+        kwl_sec = {metadata_key: (metadata_value + delta_value)}
+
+    # Add same number of col/row
+    dim = 1000
+    kwl_ref[str(MetadataKeyInGeom.NUMBER_LINE)] = dim
+    kwl_sec[str(MetadataKeyInGeom.NUMBER_LINE)] = dim
+    kwl_ref[str(MetadataKeyInGeom.NUMBER_COL)] = dim
+    kwl_sec[str(MetadataKeyInGeom.NUMBER_COL)] = dim
+
+    yield kwl_ref, kwl_sec
+
+
+@pytest.fixture(scope="function")
+def define_block_size(gdal_cache_max_value):
+    """Get block size from gdal_cache_max value"""
+    # Set GDAL_CACHEMAX  and convert in Bytes
+    gdal.SetCacheMax((gdal_cache_max_value*1024*1024))
+
+    yield utils.define_block_size_from_gdal_cache_max()
+
+@pytest.fixture(scope="function")
+def build_resampling(mocker, create_dict_kwl, same_frequencies, dimension):
+    """Build resampler and check if a resampling is required
+    """
+    # Mock isfile function to avoid Exception on input.tiff and output.tiff
+    mocker.patch(
+            'diapotb.lib.internal_processing.core.processing_dimensions.os.path.isfile',
+            return_value = True)
+
+    # Retrive kwls
+    kwl_ref, kwl_sec = create_dict_kwl
+
+    # Get f_in and f_out from kwl
+    f_in, f_out = retrieve_frequencies(kwl_ref, kwl_sec, dimension)
+
+    # Define a block size for reading/resampling/writing
+    # Could be 512, 1024 or 2048 folowwing gdal_cachemax
+    block_size = utils.define_block_size_from_gdal_cache_max()
+
+    # Instanciate a ResamplingDimension with a fake input/output file
+    resampler = ResamplingDimension(dimension, block_size, "./input.tiff",
+                                    "./output.tiff")
+
+    # Assign f_in and f_out
+    resampler.f_in = f_in
+    resampler.f_out = f_out
+    # Dummy f0 (0 here)
+    resampler.f0 = 0
+
+    # check if a resampling is required (if frequencies are different)
+    do_resampling = resampler.do_resampling(int(kwl_sec[str(MetadataKeyInGeom.NUMBER_LINE)]),
+                                            int(kwl_sec[str(MetadataKeyInGeom.NUMBER_COL)]))
+
+    yield do_resampling
+
+
+@pytest.fixture(scope="function")
+def prepare_resampling(tmpdir_factory):
+    """Create input/output and kwl to apply a resampling
+    """
+    # Create tmp dir (with tmpdir_factory pytest fixture) and tmp_file
+    my_tmpdir = tmpdir_factory.mktemp("run_resampling")
+
+    # Get inputs from current directory and get data
+    current_dir = os.path.dirname(os.path.realpath(__file__))
+
+    data_dir = os.path.join(current_dir,
+                            "../../data/Input/")
+
+
+    # Get S4Reunion data
+    input_file = os.path.abspath(os.path.join(data_dir,
+                                              "S1_Reunion/s1b_s4_20160929_extract.tiff"))
+
+    output_file = os.path.join(my_tmpdir, "s1b_s4_20160929_extract_resamp.tiff")
+
+    # Get and read geom
+    geom_file = os.path.join(os.path.dirname(input_file),
+                             os.path.basename(input_file).split(".")[0] + ".geom")
+    kwl_sec = utils.read_geom(geom_file)
+    kwl_ref = deepcopy(kwl_sec)
+
+    # Move sightly frequency to force a resampling
+    # In range
+    kwl_ref[str(MetadataKeyInGeom.FREQ_SAMPLING)] = kwl_ref[str(MetadataKeyInGeom.FREQ_SAMPLING)] - \
+                                                    kwl_ref[str(MetadataKeyInGeom.FREQ_SAMPLING)]*0.05
+    kwl_ref[str(MetadataKeyInGeom.LINE_INTERVAL)] = kwl_ref[str(MetadataKeyInGeom.LINE_INTERVAL)] - \
+                                                    kwl_ref[str(MetadataKeyInGeom.LINE_INTERVAL)]*0.05
+
+
+    yield input_file, output_file, kwl_ref, kwl_sec
+
+    # Remove tmp_dir
+    shutil.rmtree(str(my_tmpdir))
+
+#### Tests ####
+class TestResamplerPreProcessing():
+    """ Test to check when a resampling is required and annexe functions
+    """
+    # In MB
+    @pytest.fixture(scope="class", params=[350,
+                                           500,
+                                           1000])
+    def gdal_cache_max_value(self, request):
+        return request.param
+
+    @pytest.fixture(scope="class", params=[Dimension.RANGE,
+                                           Dimension.AZIMUT])
+    def dimension(self, request):
+        return request.param
+
+    @pytest.fixture(scope="class", params=[True,
+                                           False])
+    def same_frequencies(self, request):
+        return request.param
+
+    def test_block_size(self, gdal_cache_max_value, define_block_size):
+        """Check block_size following GDAL_CACHEMAX"""
+        if gdal_cache_max_value < 400:
+            assert define_block_size == 512
+        elif gdal_cache_max_value < 800:
+            assert define_block_size == 1024
+        else:
+            assert define_block_size == 2048
+
+    def test_do_resampling(self, build_resampling, dimension, same_frequencies):
+        """Check if an resampling is required """
+
+        do_resampling = build_resampling
+        # same frequencies => no need to apply a resampling
+        if same_frequencies:
+            assert not do_resampling
+        # otherwise, a resmapling is needed
+        else:
+            assert do_resampling
+
+class TestResampler():
+    """ Test to apply a resampling
+    """
+
+    def test_run_resampling(self, prepare_resampling):
+        """Run a resampling on S1Reunion """
+        # Get input, ouput, kwl_ref and kwl_sec
+        input_file, output_file, kwl_ref, kwl_sec = prepare_resampling
+
+        # Call main_resampling
+        resample_file = main_resampling(input_file, output_file, kwl_ref, kwl_sec, 0,
+                                        os.path.dirname(output_file))
+
+        # Get resample geom
+        resample_geom_file = os.path.join(os.path.dirname(resample_file),
+                                          os.path.basename(resample_file).split(".")[0] + ".geom")
+
+        # Check files
+        assert resample_file != input_file
+        assert os.path.exists(resample_file)
+        assert os.path.exists(resample_geom_file)