Merge pull request #620 from jbouffard/bug-fix/point-values

get_point_values Will Now Return the Correct Number of Results for Temporal Layers

docs/guides/layers.rst

@@ -643,9 +643,9 @@ Getting the Point Values From a SPACETIME Layer
 '''''''''''''''''''''''''''''''''''''''''''''''
 
 When using ``get_point_values`` on a layer with a ``LayerType`` of
-``SPACETIME``, the results will be paired as ``(shapely.geometry.Point, datetime.datetime, [float])``.
-Where each given ``Point`` will be paired with the values it intersects and those
-values' corresponding timestamps.
+``SPACETIME``, the results will be paired as ``(shapely.geometry.Point, [(datetime.datetime, [float])])``.
+Where each given ``Point`` will be paired with a list of tuples that contain the values it
+intersects and those values' corresponding timestamps.
 
 .. code:: python3
 
@@ -658,7 +658,7 @@ Giving a [shapely.geometry.Point] to get_point_values
 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
 When ``points`` is given as a ``[shapely.geometry.Point]``,
-then the ouput will be a ``[(shapely.geometry.Point, datetime.datetime, [float])]``.
+then the ouput will be a ``[(shapely.geometry.Point, [(datetime.datetime, [float])])]``.
 
 .. code:: python3
 
@@ -668,7 +668,7 @@ Giving a {k: shapely.geometry.Point} to get_point_values
 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
 When ``points`` is given as a ``{k: shapely.geometry.Point}``,
-then the ouput will be a ``{k: (shapely.geometry.Point, datetime.datetime, [float])}``.
+then the ouput will be a ``{k: (shapely.geometry.Point, [(datetime.datetime, [float])])}``.
 
 .. code:: python3
 

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

@@ -424,7 +424,7 @@ class TemporalTiledRasterLayer(
   def getPointValues(
     points: java.util.Map[Long, Array[Byte]],
     resampleMethod: PointResampleMethod
-  ): java.util.Map[Long, (Long, Array[Double])] = {
+  ): java.util.Map[Long, java.util.Map[String, Array[Double]]] = {
     val mapTrans = rdd.metadata.layout.mapTransform
 
     val idedKeys: Map[Long, Point] =
@@ -451,24 +451,16 @@ class TemporalTiledRasterLayer(
         case _ => _getPointValues(pointKeys, mapTrans)
       }
 
-    val remainingKeys = idedKeys.keySet diff matchedKeys.keySet
-
-    if (remainingKeys.isEmpty)
-      matchedKeys.asJava
-    else
-      remainingKeys.foldLeft(matchedKeys){ case (acc, elem) =>
-        acc + (elem -> null)
-      }.asJava
+    matchedKeys.mapValues { _.asJava }.asJava
   }
 
   def _getPointValues(
     pointKeys: Map[SpatialKey, Array[(Long, Point)]],
     mapTrans: MapKeyTransform,
     resampleMethod: PointResampleMethod
-  ): Map[Long, (Long, Array[Double])] = {
-    val resamplePoint = (tile: Tile, extent: Extent, point: Point) => {
+  ): Map[Long, Map[String, Array[Double]]] = {
+    val resamplePoint = (tile: Tile, extent: Extent, point: Point) =>
       Resample(resampleMethod, tile, extent).resampleDouble(point)
-    }
 
     rdd.flatMap { case (k, v) =>
       pointKeys.get(k.getComponent[SpatialKey]) match {
@@ -477,17 +469,17 @@ class TemporalTiledRasterLayer(
           val rasterExtent = RasterExtent(keyExtent, v.cols, v.rows)
 
           arr.map { case (id, point) =>
-            (id -> (k.instant, v.bands.map { resamplePoint(_, keyExtent, point) } toArray))
+            (id -> Map(k.time.toString -> v.bands.map { resamplePoint(_, keyExtent, point) }.toArray ))
           }
         case None => Seq()
       }
-    }.collect().toMap
+    }.reduceByKey { case (m1, m2) => m1 ++ m2 }.collect().toMap
   }
 
   def _getPointValues(
     pointKeys: Map[SpatialKey, Array[(Long, Point)]],
     mapTrans: MapKeyTransform
-  ): Map[Long, (Long, Array[Double])] =
+  ): Map[Long, Map[String, Array[Double]]] =
     rdd.flatMap { case (k, v) =>
       pointKeys.get(k.getComponent[SpatialKey]) match {
         case Some(arr) =>
@@ -503,11 +495,11 @@ class TemporalTiledRasterLayer(
               values(index) = v.band(index).getDouble(gridCol, gridRow)
             }
 
-            (id -> (k.instant, values))
+            (id -> Map(k.time.toString -> values))
           }
         case None => Seq()
       }
-    }.collect().toMap
+    }.reduceByKey { case (m1, m2) => m1 ++ m2 }.collect().toMap
 
   def filterByTimes(
     times: java.util.ArrayList[String]

geopyspark/geotrellis/layer.py

@@ -6,6 +6,7 @@
 import json
 import datetime
 import shapely.wkb
+from dateutil import parser
 import pytz
 from shapely.geometry import Polygon, MultiPolygon, Point
 from geopyspark.geotrellis.protobufcodecs import (multibandtile_decoder,
@@ -1798,18 +1799,19 @@ def get_point_values(self, points, resample_method=None):
                 [(shapely.geometry.Point, [float])]
 
             If ``points`` is a ``list`` and the ``layer_type`` is ``SPACETIME``:
-                [(shapely.geometry.Point, datetime.datetime, [float])]
+                [(shapely.geometry.Point, [(datetime.datetime, [float])])]
 
             If ``points`` is a ``dict`` and the ``layer_type`` is ``SPATIAL``:
                 {k: (shapely.geometry.Point, [float])}
 
             If ``points`` is a ``dict`` and the ``layer_type`` is ``SPACETIME``:
-                {k: (shapely.geometry.Point, datetime.datetime, [float])}
+                {k: (shapely.geometry.Point, [(datetime.datetime, [float])])}
 
             The ``shapely.geometry.Point`` in all of these returns is the original sampled point
             given. The ``[float]`` are the sampled values, one for each band. If the ``layer_type``
             was ``SPACETIME``, then the timestamp will also be included in the results represented
-            by a ``datetime.datetime`` instance.
+            by a ``datetime.datetime`` instance. These times and their associated values will be
+            given as a list of tuples for each point.
 
             Note:
                 The sampled values will always be returned as ``float``\s. Regardless of the
@@ -1835,9 +1837,6 @@ def get_point_values(self, points, resample_method=None):
         ided_points = {}
         ided_bytes = {}
 
-        def to_datetime(instant):
-            return datetime.datetime.utcfromtimestamp(instant / 1000)
-
         if isinstance(points, list):
             list_result = []
 
@@ -1858,13 +1857,14 @@ def to_datetime(instant):
                 for key, value in scala_result.items():
                     list_result.append((ided_points[key], list(value) if value else None))
             else:
-                for key, value in scala_result.items():
-                    list_result.append(
-                        (ided_points[key],
-                         to_datetime(value._1()) if value else None,
-                         list(value._2()) if value else None
-                        )
-                    )
+                for key, value_map in scala_result.items():
+                    converted_values = [(parser.parse(k), list(v)) for k, v in value_map.items()]
+                    list_result.append((ided_points[key], converted_values))
+
+                # Due to how the values sent over for temporal layers, non-intersecting keys
+                # need to be added in Python.
+                for remaining_id in set(ided_points.keys()).difference(set(scala_result.keys())):
+                    list_result.append((ided_points[remaining_id], [(None, None)]))
 
             return list_result
 
@@ -1890,12 +1890,12 @@ def to_datetime(instant):
                 for key, value in scala_result.items():
                     dict_result[ided_labels[key]] = (ided_points[key], list(value) if value else None)
             else:
-                for key, value in scala_result.items():
-                    dict_result[ided_labels[key]] = (
-                        ided_points[key],
-                        to_datetime(value._1()) if value else None,
-                        list(value._2()) if value else None
-                    )
+                for key, value_map in scala_result.items():
+                    converted_values = [(parser.parse(k), list(v)) for k, v in value_map.items()]
+                    dict_result[ided_labels[key]] = (ided_points[key], converted_values)
+
+                for remaining_id in set(ided_points.keys()).difference(set(scala_result.keys())):
+                    dict_result[ided_labels[remaining_id]] = (ided_points[remaining_id], [(None, None)])
 
             return dict_result
         else:

geopyspark/tests/geotrellis/tiled_layer_tests/point_values_test.py

@@ -1,7 +1,7 @@
-import datetime
 import numpy as np
 import pytest
 import unittest
+from dateutil import parser
 
 from shapely.geometry import Point
 
@@ -22,7 +22,8 @@ class PointValuesTest(BaseTestClass):
     extent = {'xmin': 0.0, 'ymin': 0.0, 'xmax': 4.0, 'ymax': 4.0}
     layout = {'layoutCols': 1, 'layoutRows': 1, 'tileCols': 4, 'tileRows': 4}
 
-    now = datetime.datetime.strptime("2017-09-25T11:37:00Z", '%Y-%m-%dT%H:%M:%SZ')
+    now = parser.parse("2017-09-25T11:37:00Z")
+    then = parser.parse("2018-01-17T13:53:00Z")
 
     points = [
         Point(1.0, -3.0),
@@ -49,11 +50,11 @@ class PointValuesTest(BaseTestClass):
     ]
 
     expected_spacetime_points_list = [
-        (Point(1.0, -3.0), now, [1, 2]),
-        (Point(2.0, 4.0), now, [1, 2]),
-        (Point(3.0, 3.0), now, [1, 2]),
-        (Point(1.0, -2.0), now, [1, 2]),
-        (Point(-10.0, 15.0), None, None)
+        (Point(1.0, -3.0), [(now, [1, 2]), (then, [2, 3])]),
+        (Point(2.0, 4.0), [(now, [1, 2]), (then, [2, 3])]),
+        (Point(3.0, 3.0), [(now, [1, 2]), (then, [2, 3])]),
+        (Point(1.0, -2.0), [(now, [1, 2]), (then, [2, 3])]),
+        (Point(-10.0, 15.0), [(None, None)]),
     ]
 
     expected_spatial_points_dict = {
@@ -100,11 +101,16 @@ def create_spatial_layer(self):
     def create_spacetime_layer(self):
         cells = np.array([self.first, self.second], dtype='int')
         tile = Tile.from_numpy_array(cells, -1)
+        tile2 = Tile.from_numpy_array(cells + 1, -1)
 
         layer = [(SpaceTimeKey(0, 0, self.now), tile),
                  (SpaceTimeKey(1, 0, self.now), tile),
                  (SpaceTimeKey(0, 1, self.now), tile),
-                 (SpaceTimeKey(1, 1, self.now), tile)]
+                 (SpaceTimeKey(1, 1, self.now), tile),
+                 (SpaceTimeKey(0, 0, self.then), tile2),
+                 (SpaceTimeKey(1, 0, self.then), tile2),
+                 (SpaceTimeKey(0, 1, self.then), tile2),
+                 (SpaceTimeKey(1, 1, self.then), tile2)]
 
         rdd = BaseTestClass.pysc.parallelize(layer)
 
@@ -196,7 +202,6 @@ def test_spacetime_dict_no_resample(self):
     def test_spacetime_dict_with_resample(self):
         result = self.create_spacetime_layer().get_point_values(self.labeled_points,
                                                                 ResampleMethod.NEAREST_NEIGHBOR)
-
         self.assertEqual(len(result), len(self.expected_spacetime_points_dict))
 
         keys = self.expected_spacetime_points_dict.keys()