feat: apply a cutline if supplied (#240)

* refactor: add basic types to gdalinfo response

* feat: detect the band count in imagery

* wip: attempt to apply a cutline

* refactor: correct mypy typing

* refactor: fixup lint/isort

* feat: download tiff/cutlines if we have multiple s3 locations

* feat: basic tests for presets

* refactor: fix import sorting

* refactor add more tests

* refactor: apply formatter

* docs: add docs

* refactor: fix lint

scripts/create_stac.py

@@ -1,10 +1,10 @@
-from typing import Any, Dict, Optional
+from typing import Optional
 
 from linz_logger import get_log
 
 from scripts.files.files_helper import get_file_name_from_path
 from scripts.files.geotiff import get_extents
-from scripts.gdal.gdalinfo import gdal_info
+from scripts.gdal.gdalinfo import GdalInfo, gdal_info
 from scripts.stac.imagery.item import ImageryItem
 
 
@@ -13,7 +13,7 @@ def create_item(
     start_datetime: str,
     end_datetime: str,
     collection_id: str,
-    gdalinfo_result: Optional[Dict[Any, Any]] = None,
+    gdalinfo_result: Optional[GdalInfo] = None,
 ) -> ImageryItem:
     id_ = get_file_name_from_path(file)
 

scripts/files/file_check.py

@@ -7,7 +7,7 @@
 
 from scripts.files.files_helper import get_file_name_from_path
 from scripts.gdal.gdal_helper import GDALExecutionException, run_gdal
-from scripts.gdal.gdalinfo import gdal_info
+from scripts.gdal.gdalinfo import GdalInfo, gdal_info
 from scripts.tile.tile_index import Point, TileIndexException, get_tile_name
 
 
@@ -31,10 +31,10 @@ def __init__(
         self.scale = scale
         self.errors: List[Dict[str, Any]] = []
         self._valid = True
-        self._gdalinfo: Dict[Any, Any] = {}
+        self._gdalinfo: Optional[GdalInfo] = None
         self._srs = srs
 
-    def get_gdalinfo(self) -> Optional[Dict[Any, Any]]:
+    def get_gdalinfo(self) -> Optional[GdalInfo]:
         if self.is_error_type(FileCheckErrorType.GDAL_INFO):
             return None
         if not self._gdalinfo:
@@ -65,7 +65,7 @@ def is_error_type(self, error_type: str) -> bool:
                 return True
         return False
 
-    def check_no_data(self, gdalinfo: Dict[Any, Any]) -> None:
+    def check_no_data(self, gdalinfo: GdalInfo) -> None:
         """Add an error if there is no "noDataValue" or the "noDataValue" is not equal to 255 in the "bands"."""
         bands = gdalinfo["bands"]
         if "noDataValue" in bands[0]:
@@ -74,12 +74,12 @@ def check_no_data(self, gdalinfo: Dict[Any, Any]) -> None:
                 self.add_error(
                     error_type=FileCheckErrorType.NO_DATA,
                     error_message="noDataValue is not 255",
-                    custom_fields={"current": f"{int(current_nodata_val)}"},
+                    custom_fields={"current": f"{current_nodata_val}"},
                 )
         else:
             self.add_error(error_type=FileCheckErrorType.NO_DATA, error_message="noDataValue not set")
 
-    def check_band_count(self, gdalinfo: Dict[Any, Any]) -> None:
+    def check_band_count(self, gdalinfo: GdalInfo) -> None:
         """Add an error if there is no exactly 3 bands found."""
         bands = gdalinfo["bands"]
         bands_num = len(bands)
@@ -99,7 +99,7 @@ def check_srs(self, gdalsrsinfo_tif: bytes) -> None:
         if gdalsrsinfo_tif != self._srs:
             self.add_error(error_type=FileCheckErrorType.SRS, error_message="different srs")
 
-    def check_color_interpretation(self, gdalinfo: Dict[Any, Any]) -> None:
+    def check_color_interpretation(self, gdalinfo: GdalInfo) -> None:
         """Add an error if the colors don't match RGB.
 
         Args:
@@ -125,7 +125,7 @@ def check_color_interpretation(self, gdalinfo: Dict[Any, Any]) -> None:
                 custom_fields={"missing": f"{', '.join(missing_bands)}"},
             )
 
-    def check_tile_and_rename(self, gdalinfo: Dict[Any, Any]) -> None:
+    def check_tile_and_rename(self, gdalinfo: GdalInfo) -> None:
         origin = Point(gdalinfo["cornerCoordinates"]["upperLeft"][0], gdalinfo["cornerCoordinates"]["upperLeft"][1])
         try:
             tile_name = get_tile_name(origin, self.scale)

scripts/files/files_helper.py

@@ -10,5 +10,9 @@ def is_tiff(path: str) -> bool:
     return path.lower().endswith((".tiff", ".tif"))
 
 
+def is_vrt(path: str) -> bool:
+    return path.lower().endswith(".vrt")
+
+
 def is_json(path: str) -> bool:
     return path.lower().endswith(".json")

scripts/files/geotiff.py

@@ -1,9 +1,13 @@
-from typing import Any, Dict, List, Tuple
+from typing import Dict, List, Tuple
 
 from shapely.geometry import Polygon
 
+from scripts.gdal.gdalinfo import GdalInfo
 
-def get_extents(gdalinfo_result: Dict[Any, Any]) -> Tuple[List[List[float]], List[float]]:
+
+def get_extents(gdalinfo_result: GdalInfo) -> Tuple[Dict[str, List[float]], List[float]]:
+    if gdalinfo_result["wgs84Extent"] is None:
+        raise Exception("No WGS84 Extent was found")
 
     geometry = gdalinfo_result["wgs84Extent"]
     bbox = Polygon(geometry["coordinates"][0]).bounds

scripts/files/tests/file_check_test.py

@@ -1,13 +1,16 @@
 from scripts.files.file_check import FileCheck
+from scripts.gdal.tests.gdalinfo import add_band, fake_gdal_info
 
 
 def test_check_band_count_valid() -> None:
     """
     tests check_band_count when the input layer has a valid band count
     which is 3 bands
     """
-    gdalinfo = {}
-    gdalinfo["bands"] = [{"band": 1}, {"band": 2}, {"band": 3}]
+    gdalinfo = fake_gdal_info()
+    add_band(gdalinfo)
+    add_band(gdalinfo)
+    add_band(gdalinfo)
 
     file_check = FileCheck("test", 500, b"test")
     file_check.check_band_count(gdalinfo)
@@ -20,8 +23,9 @@ def test_check_band_count_invalid() -> None:
     tests check_band_count when the input layer has a invalid band count of 2
     which is 3 bands to be valid
     """
-    gdalinfo = {}
-    gdalinfo["bands"] = [{"band": 1}, {"band": 2}]
+    gdalinfo = fake_gdal_info()
+    add_band(gdalinfo)
+    add_band(gdalinfo)
 
     file_check = FileCheck("test", 500, b"test")
     file_check.check_band_count(gdalinfo)
@@ -33,18 +37,10 @@ def test_check_color_interpretation_valid() -> None:
     """
     tests check_color_interpretation with the correct color interpretation
     """
-    gdalinfo = {}
-    gdalinfo["bands"] = [
-        {
-            "colorInterpretation": "Red",
-        },
-        {
-            "colorInterpretation": "Green",
-        },
-        {
-            "colorInterpretation": "Blue",
-        },
-    ]
+    gdalinfo = fake_gdal_info()
+    add_band(gdalinfo, color_interpretation="Red")
+    add_band(gdalinfo, color_interpretation="Green")
+    add_band(gdalinfo, color_interpretation="Blue")
 
     file_check = FileCheck("test", 500, b"test")
     file_check.check_color_interpretation(gdalinfo)
@@ -56,21 +52,11 @@ def test_check_color_interpretation_invalid() -> None:
     """
     tests check_color_interpretation with the incorrect color interpretation
     """
-    gdalinfo = {}
-    gdalinfo["bands"] = [
-        {
-            "colorInterpretation": "Red",
-        },
-        {
-            "colorInterpretation": "Green",
-        },
-        {
-            "colorInterpretation": "Blue",
-        },
-        {
-            "colorInterpretation": "Undefined",
-        },
-    ]
+    gdalinfo = fake_gdal_info()
+    add_band(gdalinfo, color_interpretation="Red")
+    add_band(gdalinfo, color_interpretation="Green")
+    add_band(gdalinfo, color_interpretation="Blue")
+    add_band(gdalinfo, color_interpretation="undefined")
 
     file_check = FileCheck("test", 500, b"test")
     file_check.check_color_interpretation(gdalinfo)
@@ -82,12 +68,8 @@ def test_check_no_data_valid() -> None:
     """
     tests check_no_data when the input layer has a valid no data value of 255
     """
-    gdalinfo = {}
-    gdalinfo["bands"] = [
-        {
-            "noDataValue": 255,
-        }
-    ]
+    gdalinfo = fake_gdal_info()
+    add_band(gdalinfo, no_data_value=255)
 
     file_check = FileCheck("test", 500, b"test")
     file_check.check_no_data(gdalinfo)
@@ -99,8 +81,8 @@ def test_check_no_data_no_value() -> None:
     """
     tests check_no_data when the input layer has no no_data value assigned
     """
-    gdalinfo = {}
-    gdalinfo["bands"] = [{"test": 1}]
+    gdalinfo = fake_gdal_info()
+    add_band(gdalinfo)
 
     file_check = FileCheck("test", 500, b"test")
     file_check.check_no_data(gdalinfo)
@@ -112,12 +94,8 @@ def test_check_no_data_invalid_value() -> None:
     """
     tests check_no_data when the input layer has the wrong value of 0 assigned
     """
-    gdalinfo = {}
-    gdalinfo["bands"] = [
-        {
-            "noDataValue": 0,
-        }
-    ]
+    gdalinfo = fake_gdal_info()
+    add_band(gdalinfo, no_data_value=0)
 
     file_check = FileCheck("test", 500, b"test")
     file_check.check_no_data(gdalinfo)

scripts/gdal/gdal_bands.py

@@ -0,0 +1,60 @@
+from typing import List, Optional
+
+from linz_logger import get_log
+
+from scripts.gdal.gdalinfo import GdalInfo, GdalInfoBand, gdal_info
+
+
+def find_band(bands: List[GdalInfoBand], color: str) -> Optional[GdalInfoBand]:
+    """Look for a specific colorInterperation inside of a gdalinfo band output
+
+    Args:
+        bands: Bands to search
+        color: Color to search, eg Red, Green, Gray
+
+    Returns:
+       Band if it exists, None otherwise
+    """
+    for band in bands:
+        if band["colorInterpretation"] == color:
+            return band
+    return None
+
+
+def get_gdal_band_offset(file: str, info: Optional[GdalInfo] = None) -> List[str]:
+    """Get the banding parameters for a gdal_translate command
+
+    Args:
+        file: file to check
+        info: optional precomputed gdalinfo
+
+    Returns:
+        list of band mappings eg "-b 1 -b 1 -b 1"
+    """
+    if info is None:
+        info = gdal_info(file, False)
+
+    bands = info["bands"]
+
+    alpha_band = find_band(bands, "Alpha")
+    alpha_band_info: List[str] = []
+    if alpha_band:
+        alpha_band_info.extend(["-b", str(alpha_band["band"])])
+
+    # Grey scale imagery, set R,G and B to just the grey_band
+    grey_band = find_band(bands, "Gray")
+    if grey_band:
+        grey_band_index = str(grey_band["band"])
+        return ["-b", grey_band_index, "-b", grey_band_index, "-b", grey_band_index] + alpha_band_info
+
+    band_red = find_band(bands, "Red")
+    band_green = find_band(bands, "Green")
+    band_blue = find_band(bands, "Blue")
+
+    if band_red is None or band_green is None or band_blue is None:
+        get_log().warn(
+            "gdal_info_bands_failed", band_red=band_red is None, band_green=band_green is None, band_blue=band_blue is None
+        )
+        return ["-b", "1", "-b", "2", "-b", "3"] + alpha_band_info
+
+    return ["-b", str(band_red["band"]), "-b", str(band_green["band"]), "-b", str(band_blue["band"])] + alpha_band_info

scripts/gdal/gdal_helper.py

@@ -64,7 +64,7 @@ def run_gdal(
 
     Args:
         command (List[str]): each arguments of the GDAL command.
-        input_file (str, optional): the input file path.
+        input_file str: the input file path.
         output_file (str, optional): the output file path.
 
     Raises:
@@ -102,7 +102,7 @@ def run_gdal(
     if proc.stderr:
         get_log().warning("run_gdal_stderr", command=command_to_string(temp_command), stderr=proc.stderr.decode())
 
-    get_log().debug("run_gdal_succeeded", command=command_to_string(temp_command), stdout=proc.stdout.decode())
+    get_log().trace("run_gdal_succeeded", command=command_to_string(temp_command), stdout=proc.stdout.decode())
 
     return proc