[test] add raster helpers test cases

samgis/io/raster_helpers.py

@@ -1,9 +1,9 @@
 """helpers for computer vision duties"""
 import numpy as np
-from numpy import ndarray
+from numpy import ndarray, bitwise_not
 
 from samgis import app_logger
-from samgis.utilities.type_hints import TmsTerrainProvidersNames
+from samgis.utilities.type_hints import XYZTerrainProvidersNames
 
 
 def get_nextzen_terrain_rgb_formula(red: ndarray, green: ndarray, blue: ndarray) -> ndarray:
@@ -34,8 +34,8 @@ def get_mapbox__terrain_rgb_formula(red: ndarray, green: ndarray, blue: ndarray)
 
 
 providers_terrain_rgb_formulas = {
-    TmsTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME: get_mapbox__terrain_rgb_formula,
-    TmsTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME: get_nextzen_terrain_rgb_formula
+    XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME: get_mapbox__terrain_rgb_formula,
+    XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME: get_nextzen_terrain_rgb_formula
 }
 
 
@@ -84,6 +84,7 @@ def get_rgb_prediction_image(raster_cropped: ndarray, slope_cellsize: int, inver
 
     try:
         slope, curvature = get_slope_curvature(raster_cropped, slope_cellsize=slope_cellsize)
+
         channel0 = raster_cropped
         channel1 = normalize_array_list(
             [raster_cropped, slope, curvature], CHANNEL_EXAGGERATIONS_LIST, title=f"channel1_normlist")
@@ -126,7 +127,8 @@ def get_rgb_image(arr_channel0: ndarray, arr_channel1: ndarray, arr_channel2: nd
         data_rgb[:, :, 2] = normalize_array(
             arr_channel2.astype(float), high=1, norm_type="float", title=f"RGB:channel2") * 192
         if invert_image:
-            data_rgb = np.bitwise_not(data_rgb)
+            app_logger.debug(f"data_rgb:{type(data_rgb)}, {data_rgb.dtype}.")
+            data_rgb = bitwise_not(data_rgb)
         return data_rgb
     except ValueError as ve_get_rgb_image:
         msg = f"ve_get_rgb_image:{ve_get_rgb_image}."
@@ -208,6 +210,7 @@ def normalize_array(arr: ndarray, high: int = 255, norm_type: str = "float", inv
         ndarray: normalized numpy array
 
     """
+    np.seterr("raise")
 
     h_min_arr = np.nanmin(arr)
     h_arr_max = np.nanmax(arr)
@@ -217,7 +220,7 @@ def normalize_array(arr: ndarray, high: int = 255, norm_type: str = "float", inv
             f"normalize_array:: '{title}',h_min_arr:{h_min_arr},h_arr_max:{h_arr_max},h_diff:{h_diff}, dtype:{arr.dtype}.")
     except Exception as e_h_diff:
         app_logger.error(f"e_h_diff:{e_h_diff}.")
-        raise e_h_diff
+        raise ValueError(e_h_diff)
 
     if check_empty_array(arr, high) or check_empty_array(arr, h_diff):
         msg_ve = f"normalize_array::empty array '{title}',h_min_arr:{h_min_arr},h_arr_max:{h_arr_max},h_diff:{h_diff}, dtype:{arr.dtype}."

samgis/io/wrappers_helpers.py

@@ -5,7 +5,7 @@ from xyzservices import providers, TileProvider
 from samgis import app_logger
 from samgis.io.coordinates_pixel_conversion import get_latlng_to_pixel_coordinates
 from samgis.utilities.constants import COMPLETE_URL_TILES_MAPBOX, COMPLETE_URL_TILES_NEXTZEN, CUSTOM_RESPONSE_MESSAGES
-from samgis.utilities.type_hints import ApiRequestBody, ContentTypes, TmsTerrainProvidersNames, TmsDefaultProvidersNames
+from samgis.utilities.type_hints import ApiRequestBody, ContentTypes, XYZTerrainProvidersNames, XYZDefaultProvidersNames
 from samgis.utilities.utilities import base64_decode
 
 
@@ -153,12 +153,12 @@ def get_parsed_request_body(event: Dict or str) -> ApiRequestBody:
 
 
 mapbox_terrain_rgb = TileProvider(
-    name=TmsTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME,
+    name=XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME,
     url=COMPLETE_URL_TILES_MAPBOX,
     attribution=""
 )
 nextzen_terrain_rgb = TileProvider(
-    name=TmsTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME,
+    name=XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME,
     url=COMPLETE_URL_TILES_NEXTZEN,
     attribution=""
 )
@@ -167,11 +167,11 @@ nextzen_terrain_rgb = TileProvider(
 def get_url_tile(source_type: str):
     try:
         match source_type.lower():
-            case TmsDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT:
-                return providers.query_name(TmsDefaultProvidersNames.DEFAULT_TILES_NAME)
-            case TmsTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME:
+            case XYZDefaultProvidersNames.DEFAULT_TILES_NAME_SHORT:
+                return providers.query_name(XYZDefaultProvidersNames.DEFAULT_TILES_NAME)
+            case XYZTerrainProvidersNames.MAPBOX_TERRAIN_TILES_NAME:
                 return mapbox_terrain_rgb
-            case TmsTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME:
+            case XYZTerrainProvidersNames.NEXTZEN_TERRAIN_TILES_NAME:
                 app_logger.info("nextzen_terrain_rgb:", nextzen_terrain_rgb)
                 return nextzen_terrain_rgb
 
@@ -184,4 +184,4 @@ def get_url_tile(source_type: str):
 
 
 def check_source_type_is_terrain(source: str | TileProvider):
-    return isinstance(source, TileProvider) and source.name in list(TmsTerrainProvidersNames)
+    return isinstance(source, TileProvider) and source.name in list(XYZTerrainProvidersNames)

samgis/prediction_api/predictors.py

@@ -7,10 +7,9 @@ from samgis.io.raster_helpers import get_raster_terrain_rgb_like, get_rgb_predic
 from samgis.io.tms2geotiff import download_extent
 from samgis.io.wrappers_helpers import check_source_type_is_terrain
 from samgis.prediction_api.sam_onnx import SegmentAnythingONNX
-from samgis.utilities.constants import MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_URL_TILES, SLOPE_CELLSIZE, \
-    DEFAULT_INPUT_SHAPE
-from samgis.utilities.type_hints import llist_float, dict_str_int, list_dict, tuple_ndarr_int, PIL_Image, \
-    TmsTerrainProvidersNames
+from samgis.utilities.constants import (
+    MODEL_ENCODER_NAME, MODEL_DECODER_NAME, DEFAULT_URL_TILES, SLOPE_CELLSIZE, DEFAULT_INPUT_SHAPE)
+from samgis.utilities.type_hints import llist_float, dict_str_int, list_dict, tuple_ndarr_int, PIL_Image
 
 models_dict = {"fastsam": {"instance": None}}
 

samgis/utilities/type_hints.py

@@ -23,13 +23,13 @@ PIL_Image = Image
 tuple_ndarray_transform = tuple[ndarray, Affine]
 
 
-class TmsDefaultProvidersNames(StrEnum):
+class XYZDefaultProvidersNames(StrEnum):
     """Default xyz provider names"""
     DEFAULT_TILES_NAME_SHORT = "openstreetmap"
     DEFAULT_TILES_NAME = "openstreetmap.mapnik"
 
 
-class TmsTerrainProvidersNames(StrEnum):
+class XYZTerrainProvidersNames(StrEnum):
     """Custom xyz provider names for digital elevation models"""
     MAPBOX_TERRAIN_TILES_NAME = "mapbox.terrain-rgb"
     NEXTZEN_TERRAIN_TILES_NAME = "nextzen.terrarium"

tests/io/test_raster_helpers.py

@@ -0,0 +1,255 @@
+import numpy as np
+import unittest
+from unittest.mock import patch
+
+from samgis.io import raster_helpers
+from samgis.utilities.utilities import hash_calculate
+
+
+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.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.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='