tests/io/test_raster_helpers.py

import unittest
from unittest.mock import patch
import numpy as np

from samgis_core.utilities.utilities import hash_calculate
from samgis_lisa_on_cuda.io import raster_helpers


def get_three_channels(size=5, param1=1000, param2=3, param3=-88):
    arr_base = np.arange(size*size).reshape(size, size) / size**2
    channel_0 = arr_base * param1
    channel_1 = arr_base * param2
    channel_2 = arr_base * param3
    return channel_0, channel_1, channel_2


def helper_bell(size=10, param1=0.1, param2=2):
    x = np.linspace(-size, size, num=size**2)
    y = np.linspace(-size, size, num=size**2)
    x, y = np.meshgrid(x, y)
    return np.exp(-param1 * x ** param2 - param1 * y ** param2)


arr_5x5x5 = np.arange(125).reshape((5, 5, 5)) / 25
arr = np.arange(25).resize((5, 5))
channel0, channel1, channel2 = get_three_channels()
z = helper_bell()
slope_z_cellsize3, curvature_z_cellsize3 = raster_helpers.get_slope_curvature(z, slope_cellsize=3)


class Test(unittest.TestCase):

    def test_get_rgb_prediction_image_real(self):
        output = raster_helpers.get_rgb_prediction_image(z, slope_cellsize=61, invert_image=True)
        hash_output = hash_calculate(output)
        assert hash_output == b'QpQ9yxgCLw9cf3klNFKNFXIDHaSkuiZxkbpeQApR8pA='
        output = raster_helpers.get_rgb_prediction_image(z, slope_cellsize=61, invert_image=False)
        hash_output = hash_calculate(output)
        assert hash_output == b'Y+iXO9w/sKzNVOw2rBh2JrVGJUFRqaa8/0F9hpevmLs='

    @patch.object(raster_helpers, "get_slope_curvature")
    @patch.object(raster_helpers, "normalize_array_list")
    @patch.object(raster_helpers, "get_rgb_image")
    def test_get_rgb_prediction_image_mocked(self, get_rgb_image_mocked, normalize_array_list, get_slope_curvature):
        local_arr = np.array(z * 100, dtype=np.uint8)

        get_slope_curvature.return_value = slope_z_cellsize3, curvature_z_cellsize3
        normalize_array_list.side_effect = None
        get_rgb_image_mocked.return_value = np.bitwise_not(local_arr)
        output = raster_helpers.get_rgb_prediction_image(local_arr, slope_cellsize=61, invert_image=True)
        hash_output = hash_calculate(output)
        assert hash_output == b'BPIyVH64RgVunj42EuQAx4/v59Va8ZAjcMnuiGNqTT0='
        get_rgb_image_mocked.return_value = local_arr
        output = raster_helpers.get_rgb_prediction_image(local_arr, slope_cellsize=61, invert_image=False)
        hash_output = hash_calculate(output)
        assert hash_output == b'XX54sdLQQUrhkUHT6ikQZYSloMYDSfh/AGITDq6jnRM='

    @patch.object(raster_helpers, "get_slope_curvature")
    def test_get_rgb_prediction_image_value_error(self, get_slope_curvature):
        msg = "this is a value error"
        get_slope_curvature.side_effect = ValueError(msg)

        with self.assertRaises(ValueError):
            try:
                raster_helpers.get_rgb_prediction_image(arr, slope_cellsize=3)
            except ValueError as ve:
                self.assertEqual(str(ve), msg)
                raise ve

    def test_get_rgb_image(self):
        output = raster_helpers.get_rgb_image(channel0, channel1, channel2, invert_image=True)
        hash_output = hash_calculate(output)
        assert hash_output == b'YVnRWla5Ptfet6reSfM+OEIsGytLkeso6X+CRs34YHk='
        output = raster_helpers.get_rgb_image(channel0, channel1, channel2, invert_image=False)
        hash_output = hash_calculate(output)
        assert hash_output == b'LC/kIZGUZULSrwwSXCeP1My2spTZdW9D7LH+tltwERs='

    def test_get_rgb_image_value_error_1(self):
        with self.assertRaises(ValueError):
            try:
                raster_helpers.get_rgb_image(arr_5x5x5, arr_5x5x5, arr_5x5x5, invert_image=True)
            except ValueError as ve:
                self.assertEqual(f"arr_size, wrong type:{type(arr_5x5x5)} or arr_size:{arr_5x5x5.shape}.", str(ve))
                raise ve

    def test_get_rgb_image_value_error2(self):
        arr_0 = np.arange(25).reshape((5, 5))
        arr_1 = np.arange(4).reshape((2, 2))
        with self.assertRaises(ValueError):
            try:
                raster_helpers.get_rgb_image(arr_0, arr_1, channel2, invert_image=True)
            except ValueError as ve:
                self.assertEqual('could not broadcast input array from shape (2,2) into shape (5,5)', str(ve))
                raise ve

    def test_get_slope_curvature(self):
        slope_output, curvature_output = raster_helpers.get_slope_curvature(z, slope_cellsize=3)
        hash_curvature = hash_calculate(curvature_output)
        hash_slope = hash_calculate(slope_output)
        assert hash_curvature == b'LAL9JFOjJP9D6X4X3fVCpnitx9VPM9drS5YMHwMZ3iE='
        assert hash_slope == b'IYf6x4G0lmR47j6HRS5kUYWdtmimhLz2nak8py75nwc='

    def test_get_slope_curvature_value_error(self):
        from samgis_lisa_on_cuda.io import raster_helpers

        with self.assertRaises(ValueError):
            try:
                raster_helpers.get_slope_curvature(np.array(1), slope_cellsize=3)
            except ValueError as ve:
                self.assertEqual('not enough values to unpack (expected 2, got 0)', str(ve))
                raise ve

    def test_calculate_slope(self):
        slope_output = raster_helpers.calculate_slope(z, cell_size=3)
        hash_output = hash_calculate(slope_output)
        assert hash_output == b'IYf6x4G0lmR47j6HRS5kUYWdtmimhLz2nak8py75nwc='

    def test_calculate_slope_value_error(self):
        with self.assertRaises(ValueError):
            try:
                raster_helpers.calculate_slope(np.array(1), cell_size=3)
            except ValueError as ve:
                self.assertEqual('not enough values to unpack (expected 2, got 0)', str(ve))
                raise ve

    def test_normalize_array(self):
        def check_ndarrays_almost_equal(cls, arr1, arr2, places, check_type="float", check_ndiff=1):
            count_abs_diff = 0
            for list00, list01 in zip(arr1.tolist(), arr2.tolist()):
                for el00, el01 in zip(list00, list01):
                    ndiff = abs(el00 - el01)
                    if el00 != el01:
                        count_abs_diff += 1
                    if check_type == "float":
                        cls.assertAlmostEqual(el00, el01, places=places)
                    cls.assertTrue(ndiff < check_ndiff)
            print("count_abs_diff:", count_abs_diff)

        normalized_array = raster_helpers.normalize_array(z)
        hash_output = hash_calculate(normalized_array)
        assert hash_output == b'MPkQwiiQa5NxL7LDvCS9V143YUEJT/Qh1aNEKc/Ehvo='

        mult_variable = 3.423
        test_array_input = np.arange(256).reshape((16, 16))
        test_array_output = raster_helpers.normalize_array(test_array_input * mult_variable)
        check_ndarrays_almost_equal(self, test_array_output, test_array_input, places=8)

        test_array_output1 = raster_helpers.normalize_array(test_array_input * mult_variable, high=128, norm_type="int")
        o = np.arange(256).reshape((16, 16)) / 2
        expected_array_output1 = o.astype(int)
        check_ndarrays_almost_equal(
            self, test_array_output1, expected_array_output1, places=2, check_type="int", check_ndiff=2)

    @patch.object(np, "nanmin")
    @patch.object(np, "nanmax")
    def test_normalize_array_floating_point_error_mocked(self, nanmax_mocked, nanmin_mocked):
        nanmax_mocked.return_value = 100
        nanmin_mocked.return_value = 100

        with self.assertRaises(ValueError):
            try:
                raster_helpers.normalize_array(
                    np.arange(25).reshape((5, 5))
                )
            except ValueError as ve:
                self.assertEqual(
                    "normalize_array:::h_arr_max:100,h_min_arr:100,fe:divide by zero encountered in divide.",
                    str(ve)
                )
                raise ve

    @patch.object(np, "nanmin")
    @patch.object(np, "nanmax")
    def test_normalize_array_exception_error_mocked(self, nanmax_mocked, nanmin_mocked):
        nanmax_mocked.return_value = 100
        nanmin_mocked.return_value = np.NaN

        with self.assertRaises(ValueError):
            try:
                raster_helpers.normalize_array(
                    np.arange(25).reshape((5, 5))
                )
            except ValueError as ve:
                self.assertEqual("cannot convert float NaN to integer", str(ve))
                raise ve

    def test_normalize_array_value_error(self):
        with self.assertRaises(ValueError):
            try:
                raster_helpers.normalize_array(
                    np.zeros((5, 5))
                )
            except ValueError as ve:
                self.assertEqual(
                    "normalize_array::empty array '',h_min_arr:0.0,h_arr_max:0.0,h_diff:0.0, " 'dtype:float64.',
                    str(ve)
                )
                raise ve

    def test_normalize_array_list(self):
        normalized_array = raster_helpers.normalize_array_list([channel0, channel1, channel2])
        hash_output = hash_calculate(normalized_array)
        assert hash_output == b'+6IbhIpyb3vPElTgqqPkQdIR0umf4uFP2c7t5IaBVvI='

        test_norm_list_output2 = raster_helpers.normalize_array_list(
            [channel0, channel1, channel2], exaggerations_list=[2.0, 3.0, 5.0])
        hash_variable2 = hash_calculate(test_norm_list_output2)
        assert hash_variable2 == b'yYCYWCKO3i8NYsWk/wgYOzSRRLSLUprEs7mChJkdL+A='

    def test_normalize_array_list_value_error(self):
        with self.assertRaises(ValueError):
            try:
                raster_helpers.normalize_array_list([])
            except ValueError as ve:
                self.assertEqual("input list can't be empty:[].", str(ve))
                raise ve

    def test_check_empty_array(self):
        a = np.zeros((10, 10))
        b = np.ones((10, 10))
        c = np.ones((10, 10)) * 2
        d = np.zeros((10, 10))
        d[1, 1] = np.nan
        e = np.ones((10, 10)) * 3
        e[1, 1] = np.nan

        self.assertTrue(raster_helpers.check_empty_array(a, 999))
        self.assertTrue(raster_helpers.check_empty_array(b, 0))
        self.assertTrue(raster_helpers.check_empty_array(c, 2))
        self.assertTrue(raster_helpers.check_empty_array(d, 0))
        self.assertTrue(raster_helpers.check_empty_array(e, 3))
        self.assertFalse(raster_helpers.check_empty_array(z, 3))

    def test_get_nextzen_terrain_rgb_formula(self):
        output = raster_helpers.get_nextzen_terrain_rgb_formula(channel0, channel1, channel2)
        hash_output = hash_calculate(output)
        assert hash_output == b'3KJ81YKmQRdccRZARbByfwo1iMVLj8xxz9mfsWki/qA='

    def test_get_mapbox__terrain_rgb_formula(self):
        output = raster_helpers.get_mapbox__terrain_rgb_formula(channel0, channel1, channel2)
        hash_output = hash_calculate(output)
        assert hash_output == b'RU7CcoKoR3Fkh5LE+m48DHRVUy/vGq6UgfOFUMXx07M='

    def test_get_raster_terrain_rgb_like(self):
        from samgis_lisa_on_cuda.utilities.type_hints import XYZTerrainProvidersNames

        arr_input = raster_helpers.get_rgb_image(channel0, channel1, channel2, invert_image=True)
        output_nextzen = raster_helpers.get_raster_terrain_rgb_like(
            arr_input, XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME)
        hash_nextzen = hash_calculate(output_nextzen)
        assert hash_nextzen == b'+o2OTJliJkkBoqiAIGnhJ4s0xoLQ4MxHOvevLhNxysE='
        output_mapbox = raster_helpers.get_raster_terrain_rgb_like(
            arr_input, XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME)
        hash_mapbox = hash_calculate(output_mapbox)
        assert hash_mapbox == b'zWmekyKrpnmHnuDACnveCJl+o4GuhtHJmGlRDVwsce4='