Add filepath_per_band when for no temporal dimension. Open-EO/openeo-…

…geopyspark-driver#877

openeo-geotrellis/src/main/scala/org/openeo/geotrellis/geotiff/GTiffOptions.scala

@@ -6,19 +6,43 @@ import geotrellis.raster.render.{ColorMap, DoubleColorMap, IndexedColorMap}
 
 import scala.collection.JavaConverters._
 
+//noinspection ScalaUnusedSymbol
 class GTiffOptions extends Serializable {
 
   var filenamePrefix = "openEO" // Example using default prefix: "openEO_2017-01-02Z.tif"
   var colorMap: Option[ColorMap] = Option.empty
+  var filepathPerBand: Option[util.ArrayList[String]] = Option.empty
   var tags: Tags = Tags.empty
   var overviews:String = "OFF"
   var resampleMethod:String = "near"
   var separateAssetPerBand = false
 
-  def setFilenamePrefix(name: String): Unit = this.filenamePrefix = name
+  def setFilenamePrefix(name: String): Unit = {
+    assertSafeToUseInFilePath(name)
+    this.filenamePrefix = name
+  }
 
   def setSeparateAssetPerBand(value: Boolean): Unit = this.separateAssetPerBand = value
 
+  def setFilepathPerBand(value: Option[util.ArrayList[String]]): Unit = {
+    if (value.isDefined) {
+      // Check in lower case because Windows does not make the distinction
+      val valueLowercase = value.get.asScala.map(_.toLowerCase).toList
+      valueLowercase.foreach(filepath => {
+        assertSafeToUseInFilePath(filepath)
+
+        if (!filepath.endsWith(".tiff") && !filepath.endsWith(".tif")) {
+          throw new IllegalArgumentException("File name must end with .tiff or .tif: " + filepath)
+        }
+      })
+
+      if (valueLowercase.size != valueLowercase.distinct.size) {
+        throw new IllegalArgumentException("All paths in 'filepath_per_band' must be unique: " + value)
+      }
+    }
+    this.filepathPerBand = value
+  }
+
   def setColorMap(colors: util.ArrayList[Int]): Unit = {
     colorMap = Some(new IndexedColorMap(colors.asScala))
   }
@@ -78,4 +102,19 @@ class GTiffOptions extends Serializable {
     val ois = new java.io.ObjectInputStream(bais)
     ois.readObject().asInstanceOf[GTiffOptions]
   }
+
+  def assertNoConflicts(): Unit = {
+    if (filepathPerBand.isDefined) {
+      if (!separateAssetPerBand) {
+        throw new IllegalArgumentException("filepath_per_band is only supported with separate_asset_per_band.")
+      }
+      if (filenamePrefix != "openEO") { // Compare with default value
+        throw new IllegalArgumentException("filepath_per_band is not supported with filename_prefix.")
+      }
+      if (tags.bandCount != filepathPerBand.get.size()) {
+        throw new IllegalArgumentException("filepath_per_band size does not match the number of bands. " +
+          s"${tags.bandCount} != ${filepathPerBand.get.size()}.")
+      }
+    }
+  }
 }

openeo-geotrellis/src/main/scala/org/openeo/geotrellis/geotiff/package.scala

@@ -93,6 +93,7 @@ package object geotiff {
                       formatOptions: GTiffOptions = new GTiffOptions
                      ): java.util.List[(String, String, Extent)] = {
     rdd.sparkContext.setCallSite(s"save_result(GTiff, temporal)")
+    formatOptions.assertNoConflicts()
     val ret = saveRDDTemporalAllowAssetPerBand(rdd, path, zLevel, cropBounds, formatOptions).asScala
     logger.warn("Calling backwards compatibility version for saveRDDTemporalConsiderAssetPerBand")
     //    val duplicates = ret.groupBy(_._2).filter(_._2.size > 1)
@@ -117,6 +118,7 @@ package object geotiff {
                                           cropBounds: Option[Extent] = Option.empty[Extent],
                                           formatOptions: GTiffOptions = new GTiffOptions
                                          ): java.util.List[(String, String, Extent, java.util.List[Int])] = {
+    formatOptions.assertNoConflicts()
     val preProcessResult: (GridBounds[Int], Extent, RDD[(SpaceTimeKey, MultibandTile)] with Metadata[TileLayerMetadata[SpaceTimeKey]]) = preProcess(rdd,cropBounds)
     val gridBounds: GridBounds[Int] = preProcessResult._1
     val croppedExtent: Extent = preProcessResult._2
@@ -214,6 +216,7 @@ package object geotiff {
                                cropBounds: Option[Extent] = Option.empty[Extent],
                                formatOptions: GTiffOptions = new GTiffOptions
                               ): java.util.List[(String, java.util.List[Int])] = {
+    formatOptions.assertNoConflicts()
     if (formatOptions.separateAssetPerBand) {
       val bandLabels = formatOptions.tags.bandTags.map(_("DESCRIPTION"))
       val layout = rdd.metadata.layout
@@ -227,7 +230,11 @@ package object geotiff {
           tile =>
             bandIndex += 1
             val t = _root_.geotrellis.raster.MultibandTile(Seq(tile))
-            val name = formatOptions.filenamePrefix + "_" + bandLabels(bandIndex) + ".tif"
+            // match on formatOptions.filepathPerBand:
+            val name = formatOptions.filepathPerBand match {
+              case Some(filepathPerBand) => filepathPerBand.get(bandIndex)
+              case None => formatOptions.filenamePrefix + "_" + bandLabels(bandIndex) + ".tif"
+            }
             ((name, bandIndex), (key, t))
         }
       }
@@ -239,11 +246,16 @@ package object geotiff {
           else {
             path
           }
-
+        Path.of(fixedPath).toFile.getParentFile.mkdirs()
         val fo = formatOptions.deepClone()
         // Keep only one band tag
         val newBandTags = List(formatOptions.tags.bandTags(bandIndex))
         fo.setBandTags(newBandTags)
+        if (formatOptions.filepathPerBand.isDefined) {
+          fo.setFilepathPerBand(Some(new ArrayList[String](Collections.singletonList(
+            formatOptions.filepathPerBand.get.get(bandIndex)
+          ))))
+        }
 
         (stitchAndWriteToTiff(tiles, fixedPath, layout, crs, extent, None, None, compression, Some(fo)),
           Collections.singletonList(bandIndex))
@@ -734,6 +746,7 @@ package object geotiff {
         fo.overviews = "ALL"
         fo
     }
+    fo.assertNoConflicts()
     var geotiff = MultibandGeoTiff(adjusted.tile, adjusted.extent, crs,
       fo.tags, GeoTiffOptions(compression))
     val gridBounds = adjusted.extent
@@ -907,4 +920,33 @@ package object geotiff {
     result.collect()
     print("test done")
   }
+
+  def assertSafeToUseInFilePath(filepath: String): Unit = {
+    val name = filepath.split("/").last
+    assertValidWindowsFilename(name)
+    if (filepath.contains("..") || filepath.contains("%") || filepath.contains("|")) {
+      throw new IllegalArgumentException("Invalid filepath: " + filepath)
+    }
+  }
+
+
+  /**
+   * http://msdn.microsoft.com/en-us/library/aa365247.aspx
+   */
+  def assertValidWindowsFilename(filename: String): Unit = {
+    // TODO: Is there a standard library function for this?
+    val filenameLower = filename.toLowerCase
+    val invalidCharacters = Seq("<", ">", ":", "\"", "/", "\\", "|", "?", "*")
+    if (invalidCharacters.exists(filenameLower.contains)) {
+      throw new IllegalArgumentException("Invalid characters in filename: " + filename)
+    }
+
+    val filenameWithoutExtension = filename.split('.').head
+    val invalidNames = Seq("CON", "PRN", "AUX", "NUL", "COM0", "COM1", "COM2",
+      "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2",
+      "LPT3", "LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9")
+    if (invalidNames.contains(filenameWithoutExtension.toUpperCase())) {
+      throw new IllegalArgumentException("Invalid filename: " + filename)
+    }
+  }
 }

openeo-geotrellis/src/test/scala/org/openeo/geotrellis/geotiff/WriteRDDToGeotiffTest.scala

@@ -311,7 +311,7 @@ class WriteRDDToGeotiffTest {
     val (imageTile: ByteArrayTile, filtered: MultibandTileLayerRDD[SpatialKey]) = LayerFixtures.createLayerWithGaps(layoutCols, layoutRows)
 
     val outDir = Paths.get("tmp/testWriteMultibandRDDWithGapsSeparateAssetPerBand/")
-    new Directory(outDir.toFile).deepFiles.foreach(_.delete())
+    new Directory(outDir.toFile).deepList().foreach(_.delete())
     Files.createDirectories(outDir)
 
     val filename = outDir + "/out"
@@ -339,6 +339,46 @@ class WriteRDDToGeotiffTest {
     assertArrayEquals(croppedReference.toArray(), croppedOutput.toArray())
   }
 
+  @Test
+  def testWriteMultibandRDDWithGapsFilepathPerBand(): Unit = {
+    val layoutCols = 8
+    val layoutRows = 4
+    val (imageTile: ByteArrayTile, filtered: MultibandTileLayerRDD[SpatialKey]) = LayerFixtures.createLayerWithGaps(layoutCols, layoutRows)
+
+    val outDir = Paths.get("tmp/testWriteMultibandRDDWithGapsFilepathPerBand/")
+    new Directory(outDir.toFile).deepList().foreach(_.delete())
+    Files.createDirectories(outDir)
+
+    val filename = outDir + "/out"
+    val options = new GTiffOptions()
+    options.separateAssetPerBand = true
+
+    val filepathPerBand: util.ArrayList[String] = new util.ArrayList[String]()
+    filepathPerBand.add("testA/B01.tiff")
+    filepathPerBand.add("testA/A/B02.tiff")
+    filepathPerBand.add("testB/B03.tiff")
+    options.setFilepathPerBand(Some(filepathPerBand))
+    options.addBandTag(0, "DESCRIPTION", "B01")
+    options.addBandTag(1, "DESCRIPTION", "B02")
+    options.addBandTag(2, "DESCRIPTION", "B03")
+    val paths = saveRDD(filtered.withContext {
+      _.repartition(layoutCols * layoutRows)
+    }, 3, filename, formatOptions = options)
+    assertEquals(3, paths.size())
+
+    GeoTiff.readMultiband(outDir.resolve("testA/B01.tiff").toString).raster.tile
+    GeoTiff.readMultiband(outDir.resolve("testA/A/B02.tiff").toString).raster.tile
+    GeoTiff.readMultiband(outDir.resolve("testB/B03.tiff").toString).raster.tile
+
+    val result = GeoTiff.readMultiband(paths.get(0)).raster.tile
+
+    //crop away the area where data was removed, and check if rest of geotiff is still fine
+    val croppedReference = imageTile.crop(2 * 256, 0, layoutCols * 256, layoutRows * 256).toArrayTile()
+
+    val resultWidth = result.band(0).toArrayTile().dimensions.cols
+    val croppedOutput = result.band(0).toArrayTile().crop(resultWidth - (6 * 256), 0, layoutCols * 256, layoutRows * 256)
+    assertArrayEquals(croppedReference.toArray(), croppedOutput.toArray())
+  }
 
   @Test
   def testWriteMultibandTemporalRDDWithGaps(): Unit = {