Merge pull request #70 from WikiWatershed/tt/collections-api-futures

Collections API: Use Futures Connects #67
WikiWatershed · Sep 1, 2017 · 821d107 · 821d107
2 parents f3241af + c37560f
commit 821d107
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 18 deletions.
diff --git a/api/src/main/scala/Geoprocessing.scala b/api/src/main/scala/Geoprocessing.scala
@@ -1,6 +1,8 @@
 package org.wikiwatershed.mmw.geoprocessing
 
 import java.util.concurrent.atomic.{LongAdder, DoubleAdder}
+import scala.concurrent.Future
+import scala.concurrent.ExecutionContext.Implicits.global
 
 import collection.concurrent.TrieMap
 
@@ -17,10 +19,13 @@ trait Geoprocessing extends Utils {
     * @param   input  The InputData
     * @return         A histogram of results
     */
-  def getRasterGroupedCount(input: InputData): ResultInt = {
+  def getRasterGroupedCount(input: InputData): Future[ResultInt] = {
     val aoi = createAOIFromInput(input)
-    val rasterLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
-    ResultInt(rasterGroupedCount(rasterLayers, aoi))
+    val futureLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
+
+    futureLayers.map { layers =>
+      ResultInt(rasterGroupedCount(layers, aoi))
+    }
   }
 
   /**
@@ -30,21 +35,23 @@ trait Geoprocessing extends Utils {
     * @param   input  The InputData
     * @return         A histogram of results
     */
-  def getRasterGroupedAverage(input: InputData): ResultDouble = {
+  def getRasterGroupedAverage(input: InputData): Future[ResultDouble] = {
     val aoi = createAOIFromInput(input)
-    val rasterLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
+    val futureLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
     val targetLayer = input.targetRaster match {
       case Some(targetRaster) =>
         cropSingleRasterToAOI(targetRaster, input.zoom, aoi)
       case None =>
         throw new Exception("Request data missing required 'targetRaster'.")
     }
 
-    val average =
-      if (rasterLayers.isEmpty) rasterAverage(targetLayer, aoi)
-      else rasterGroupedAverage(rasterLayers, targetLayer, aoi)
+    futureLayers.map { rasterLayers =>
+      val average =
+        if (rasterLayers.isEmpty) rasterAverage(targetLayer, aoi)
+        else rasterGroupedAverage(rasterLayers, targetLayer, aoi)
 
-    ResultDouble(average)
+      ResultDouble(average)
+    }
   }
 
   /**
@@ -53,9 +60,9 @@ trait Geoprocessing extends Utils {
     * @param   input  The InputData
     * @return         A histogram of results
     */
-  def getRasterLinesJoin(input: InputData): ResultInt = {
+  def getRasterLinesJoin(input: InputData): Future[ResultInt] = {
     val aoi = createAOIFromInput(input)
-    val rasterLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
+    val futureLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
     val lines = input.vector match {
       case Some(vector) =>
         input.vectorCRS match {
@@ -67,7 +74,10 @@ trait Geoprocessing extends Utils {
       case None =>
         throw new Exception("Request data missing required 'vector'.")
     }
-    ResultInt(rasterLinesJoin(rasterLayers, lines))
+
+    futureLayers.map { rasterLayers =>
+      ResultInt(rasterLinesJoin(rasterLayers, lines))
+    }
   }
 
   private case class TilePixel(key: SpatialKey, col: Int, row: Int)
@@ -85,7 +95,7 @@ trait Geoprocessing extends Utils {
     lines: Seq[MultiLine]
   ): Map[String, Int] = {
     val metadata = rasterLayers.head.metadata
-    var pixelGroups: TrieMap[(List[Int], TilePixel), Int] = TrieMap.empty
+    val pixelGroups: TrieMap[(List[Int], TilePixel), Int] = TrieMap.empty
 
     joinCollectionLayers(rasterLayers).par
       .foreach({ case (key, tiles) =>
@@ -213,7 +223,7 @@ trait Geoprocessing extends Utils {
     // assume all the layouts are the same
     val metadata = rasterLayers.head.metadata
 
-    var pixelGroups: TrieMap[List[Int], LongAdder] = TrieMap.empty
+    val pixelGroups: TrieMap[List[Int], LongAdder] = TrieMap.empty
 
     joinCollectionLayers(rasterLayers).par
       .foreach({ case (key, tiles) =>

diff --git a/api/src/main/scala/Utils.scala b/api/src/main/scala/Utils.scala
@@ -1,12 +1,13 @@
 package org.wikiwatershed.mmw.geoprocessing
 
+import scala.concurrent.Future
+import scala.concurrent.ExecutionContext.Implicits.global
+
 import spray.json._
-import spray.json.DefaultJsonProtocol._
 
 import geotrellis.proj4.{CRS, ConusAlbers, LatLng, WebMercator}
 
 import geotrellis.raster._
-import geotrellis.raster.rasterize._
 import geotrellis.vector._
 import geotrellis.vector.io._
 import geotrellis.spark._
@@ -32,8 +33,10 @@ trait Utils {
     rasterIds: List[String],
     zoom: Int,
     aoi: MultiPolygon
-  ): Seq[TileLayerCollection[SpatialKey]] =
-    rasterIds.map { str => cropSingleRasterToAOI(str, zoom, aoi) }
+  ): Future[Seq[TileLayerCollection[SpatialKey]]] =
+    Future.sequence {
+      rasterIds.map { str => Future(cropSingleRasterToAOI(str, zoom, aoi))}
+    }
 
   /**
     * Given a zoom level & area of interest, transform a raster filename into a