Address PR review comments

Signed-off-by: jpolchlo <jpolchlopek@azavea.com>
locationtech · Dec 20, 2017 · 96b7b2c · 96b7b2c
1 parent aea2e24
commit 96b7b2c
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 66 deletions.
diff --git a/raster/src/main/scala/geotrellis/raster/reproject/RasterRegionReproject.scala b/raster/src/main/scala/geotrellis/raster/reproject/RasterRegionReproject.scala
@@ -27,7 +27,7 @@ trait RasterRegionReproject[T <: CellGrid] extends Serializable {
 }
 
 object RasterRegionReproject {
-  def rowCoords(destRegion: Polygon, destRasterExtent: RasterExtent, toSrcCrs: Transform): Int => (Array[Int], Array[Double], Array[Double]) = {
+  private def rowCoords(destRegion: Polygon, destRasterExtent: RasterExtent, toSrcCrs: Transform): Int => (Array[Int], Array[Double], Array[Double]) = {
     val extent = destRasterExtent.extent
     val rowTransform = RowTransform.approximate(toSrcCrs, 0.125)
 
@@ -94,24 +94,7 @@ object RasterRegionReproject {
   implicit val singlebandInstance = new RasterRegionReproject[Tile] {
     def regionReproject(raster: Raster[Tile], src: CRS, dest: CRS, rasterExtent: RasterExtent, region: Polygon, resampleMethod: ResampleMethod): Raster[Tile] = {
       val buffer = raster.tile.prototype(rasterExtent.cols, rasterExtent.rows)
-      val trans = Proj4Transform(dest, src)
-      val resampler = Resample.apply(resampleMethod, raster.tile, raster.extent, CellSize(raster.rasterExtent.cellwidth, raster.rasterExtent.cellheight))
-
-      val rowcoords = rowCoords(region, rasterExtent, trans)
-
-      cfor(0)(_ < rasterExtent.rows, _ + 1){ i =>
-        val (pxs, xs, ys) = rowcoords(i)
-        if (raster.cellType.isFloatingPoint) {
-          cfor(0)(_ < xs.size, _ + 1){ s =>
-            buffer.setDouble(pxs(s), i, resampler.resampleDouble(xs(s), ys(s)))
-          }
-        } else {
-          cfor(0)(_ < xs.size, _ + 1){ s =>
-            buffer.set(pxs(s), i, resampler.resample(xs(s), ys(s)))
-          }
-        }
-      }
-
+      mutableRegionReproject(buffer, raster, src, dest, rasterExtent, region, resampleMethod)
       Raster(buffer, rasterExtent.extent)
     }
 
@@ -139,35 +122,11 @@ object RasterRegionReproject {
 
   implicit val multibandInstance = new RasterRegionReproject[MultibandTile] {
     def regionReproject(raster: Raster[MultibandTile], src: CRS, dest: CRS, rasterExtent: RasterExtent, region: Polygon, resampleMethod: ResampleMethod): Raster[MultibandTile] = {
-      val trans = Proj4Transform(dest, src)
       val bands = Array.ofDim[MutableArrayTile](raster.tile.bandCount)
-
       cfor(0)(_ < bands.length, _ + 1) { i =>
         bands(i) = raster.band(i).prototype(rasterExtent.cols, rasterExtent.rows).mutable
       }
-
-      val resampler = (0 until raster.bandCount).map { i =>
-        Resample(resampleMethod, raster.band(i), raster.extent, raster.rasterExtent.cellSize)
-      }
-
-      if (raster.cellType.isFloatingPoint) {
-        Rasterizer.foreachCellByPolygon(region, rasterExtent) { (px, py) =>
-          val (x, y) = rasterExtent.gridToMap(px, py)
-          val (tx, ty) = trans(x, y)
-          cfor(0)(_ < bands.length, _ + 1) { i =>
-            bands(i).setDouble(px, py, resampler(i).resampleDouble(tx, ty))
-          }
-        }
-      } else {
-        Rasterizer.foreachCellByPolygon(region, rasterExtent) { (px, py) =>
-          val (x, y) = rasterExtent.gridToMap(px, py)
-          val (tx, ty) = trans(x, y)
-          cfor(0)(_ < bands.length, _ + 1) { i =>
-            bands(i).set(px, py, resampler(i).resample(tx, ty))
-          }
-        }
-      }
-
+      mutableRegionReproject(MultibandTile(bands), raster, src, dest, rasterExtent, region, resampleMethod)
       Raster(MultibandTile(bands), rasterExtent.extent)
     }
 

diff --git a/spark/src/main/scala/geotrellis/spark/reproject/TileRDDReproject.scala b/spark/src/main/scala/geotrellis/spark/reproject/TileRDDReproject.scala
@@ -193,30 +193,22 @@ object TileRDDReproject {
           }}
         }
 
-    // val inputCols = bufferedTiles.map{ case (key, BufferedTile(_, gb)) => (key.getComponent[SpatialKey].col, gb.width) }.collect.toMap.values.reduce(_+_)
-    // println(s"BufferedTile input total width: $inputCols")
-    // val outputCols = reprojectedTiles.map{ case (key, t) => (key.getComponent[SpatialKey].col, t.cols) }.collect.toMap.values.reduce(_+_)
-    // println(s"Reprojected output total width: $outputCols")
-    // val shouldBeCols = reprojectedTiles.map{ case (key, _) => (key.getComponent[SpatialKey].col, maptrans(key).width / newMetadata.layout.cellSize.width) }.collect.toMap.values.reduce(_+_)
-    // println(s"Reprojected output total width (by extent): $shouldBeCols")
-    // val expectedCols = targetDataExtent.width / newMetadata.layout.cellSize.width
-    // println(s"Expected width: $expectedCols")
-
-
-    // val tiled = reprojectedTiles.merge(part)
-    val tiled: RDD[(K, V)] = stagedTiles.combineByKey(
-      { case (raster, destRE, destRegion) =>
+    def createCombiner(tup: (Raster[V], RasterExtent, Polygon)) = {
+        val (raster, destRE, destRegion) = tup
         rrp.regionReproject(raster, crs, destCrs, destRE, destRegion, rasterReprojectOptions.method).tile
-      },
-      { (reprojectedTile, toReproject) =>
-        val (raster, destRE, destRegion) = toReproject
-        rrp.mutableRegionReproject(reprojectedTile, raster, crs, destCrs, destRE, destRegion, rasterReprojectOptions.method)
-        reprojectedTile
-      },
-      { (reproj1, reproj2) =>
-        reproj1.merge(reproj2)
       }
-    )
+    def mergeValues(reprojectedTile: V, toReproject: (Raster[V], RasterExtent, Polygon)) = {
+      val (raster, destRE, destRegion) = toReproject
+      rrp.mutableRegionReproject(reprojectedTile, raster, crs, destCrs, destRE, destRegion, rasterReprojectOptions.method)
+      reprojectedTile
+    }
+    def mergeCombiners(reproj1: V, reproj2: V) = reproj1.merge(reproj2)
+
+    val tiled: RDD[(K, V)] =
+      stagedTiles.partitioner match {
+        case Some(part) => stagedTiles.combineByKey(createCombiner, mergeValues, mergeCombiners, partitioner = part)
+        case None => stagedTiles.combineByKey(createCombiner, mergeValues, mergeCombiners)
+      }
 
     (targetZoom, ContextRDD(tiled, newMetadata))
   }