Merge pull request #377 from jamesmcclain/feature/timeseries

Time Series

Author: jamesmcclain

geopyspark-backend/geotrellis/src/main/scala/geopyspark/geotrellis/Logs.scala renamed to geopyspark-backend/geotrellis/src/main/scala/geopyspark/geotrellis/Log.scala

@@ -43,6 +43,8 @@ import org.apache.spark.rdd._
 import org.apache.spark.SparkContext._
 
 import java.util.ArrayList
+import java.time.ZonedDateTime
+
 import scala.reflect._
 import scala.collection.JavaConverters._
 
@@ -90,6 +92,78 @@ class TemporalTiledRasterLayer(
     TemporalTiledRasterLayer(zoomLevel, multiBand)
   }
 
+  private def wkbsToMultiPolygons(wkbs: java.util.ArrayList[Array[Byte]]) = {
+    wkbs
+      .asScala.map({ wkb => WKB.read(wkb) })
+      .flatMap({
+        case p: Polygon => Some(MultiPolygon(p))
+        case m: MultiPolygon => Some(m)
+        case _ => None
+      })
+  }
+
+  private def wkbsToMultiPolygon(wkbs: java.util.ArrayList[Array[Byte]]) =
+    MultiPolygon(
+      wkbsToMultiPolygons(wkbs)
+        .map({ mp => mp.polygons })
+        .foldLeft(List.empty[Polygon])(_ ++ _)
+    )
+
+  def sumSeries(
+    wkbs: java.util.ArrayList[Array[Byte]]
+  ): Array[(ZonedDateTime, Double)] = {
+    val polygon: MultiPolygon = wkbsToMultiPolygon(wkbs)
+    val metadata = rdd.metadata
+    ContextRDD(rdd.mapValues({ m => m.bands(0) }), metadata)
+      .sumSeries(polygon)
+      .toArray
+      .sortWith({ (t1, t2) => (t1._1.compareTo(t2._1) <= 0) })
+  }
+
+  def minSeries(
+    wkbs: java.util.ArrayList[Array[Byte]]
+  ): Array[(ZonedDateTime, Double)] = {
+    val polygon: MultiPolygon = wkbsToMultiPolygon(wkbs)
+    val metadata = rdd.metadata
+    ContextRDD(rdd.mapValues({ m => m.bands(0) }), metadata)
+      .minSeries(polygon)
+      .toArray
+      .sortWith({ (t1, t2) => (t1._1.compareTo(t2._1) <= 0) })
+  }
+
+  def maxSeries(
+    wkbs: java.util.ArrayList[Array[Byte]]
+  ): Array[(ZonedDateTime, Double)] = {
+    val polygon: MultiPolygon = wkbsToMultiPolygon(wkbs)
+    val metadata = rdd.metadata
+    ContextRDD(rdd.mapValues({ m => m.bands(0) }), metadata)
+      .maxSeries(polygon)
+      .toArray
+      .sortWith({ (t1, t2) => (t1._1.compareTo(t2._1) <= 0) })
+  }
+
+  def meanSeries(
+    wkbs: java.util.ArrayList[Array[Byte]]
+  ): Array[(ZonedDateTime, Double)] = {
+    val polygon: MultiPolygon = wkbsToMultiPolygon(wkbs)
+    val metadata = rdd.metadata
+    ContextRDD(rdd.mapValues({ m => m.bands(0) }), metadata)
+      .meanSeries(polygon)
+      .toArray
+      .sortWith({ (t1, t2) => (t1._1.compareTo(t2._1) <= 0) })
+  }
+
+  def histogramSeries(
+    wkbs: java.util.ArrayList[Array[Byte]]
+  ): Array[(ZonedDateTime, Histogram[Double])] = {
+    val polygon: MultiPolygon = wkbsToMultiPolygon(wkbs)
+    val metadata = rdd.metadata
+    ContextRDD(rdd.mapValues({ m => m.bands(0) }), metadata)
+      .histogramSeries(polygon)
+      .toArray
+      .sortWith({ (t1, t2) => (t1._1.compareTo(t2._1) <= 0) })
+  }
+
   def reproject(targetCRS: String, resampleMethod: ResampleMethod): TemporalTiledRasterLayer = {
     val crs = TileLayer.getCRS(targetCRS).get
     val (zoom, reprojected) = rdd.reproject(crs, rdd.metadata.layout, resampleMethod)

geopyspark-backend/geotrellis/src/main/scala/geopyspark/geotrellis/TemporalTiledRasterLayer.scala

@@ -1133,6 +1133,36 @@ def save_stitched(self, path, crop_bounds=None, crop_dimensions=None):
         else:
             self.srdd.save_stitched(path)
 
+    def star_series(self, geometries, fn):
+        if not self.layer_type == LayerType.SPACETIME:
+            raise ValueError("Only Spatio-Temporal layers can use this function.")
+
+        if not isinstance(geometries, list):
+            geometries = [geometries]
+        wkbs = [shapely.wkb.dumps(g) for g in geometries]
+
+        return [(t._1(), t._2()) for t in list(fn(wkbs))]
+
+    def histogram_series(self, geometries):
+        fn = self.srdd.histogramSeries
+        return self.star_series(geometries, fn)
+
+    def mean_series(self, geometries):
+        fn = self.srdd.meanSeries
+        return self.star_series(geometries, fn)
+
+    def max_series(self, geometries):
+        fn = self.srdd.maxSeries
+        return self.star_series(geometries, fn)
+
+    def min_series(self, geometries):
+        fn = self.srdd.minSeries
+        return self.star_series(geometries, fn)
+
+    def sum_series(self, geometries):
+        fn = self.srdd.sumSeries
+        return self.star_series(geometries, fn)
+
     def mask(self, geometries):
         """Masks the ``TiledRasterLayer`` so that only values that intersect the geometries will
         be available.