more docs and unit tests around local algebra and nodata

Signed-off-by: Jason T. Brown <[email protected]>

Author: vpipkt

pyrasterframes/src/main/python/docs/reference.pymd

@@ -238,6 +238,8 @@ If input `tile` had a NoData value already, the behaviour depends on if its cell
 
 [Local map algebra](https://gisgeography.com/map-algebra-global-zonal-focal-local/) raster operations are element-wise operations on a single tile (unary), between a `tile` and a scalar, between two `tile`s, or across many `tile`s.
 
+When these operations encounter a NoData value in either operand, the cell in the resulting `tile` will have a NoData.
+
 The binary local map algebra functions have similar variations in the Python API depending on the left hand side type:
 
  - `rf_local_op`: applies `op` to two columns; the right hand side can be a `tile` or a numeric column.
@@ -536,6 +538,8 @@ Aggregates over all of the rows in DataFrame of `tile` and returns a count of ea
 
 Local statistics compute the element-wise statistics across a DataFrame or group of `tile`s, resulting in a `tile` that has the same dimension.
 
+When these functions encounter a NoData in a cell location, it will be ignored. 
+
 ### rf_agg_local_max
 
     Tile rf_agg_local_max(Tile tile)

pyrasterframes/src/main/python/setup.py

@@ -144,7 +144,7 @@ def initialize_options(self):
         'setuptools>=0.8',
         'ipython==6.2.1',
         "ipykernel==4.8.0",
-        'Pweave==0.30.3'
+        'Pweave==0.30.3',
     ],
     tests_require=[
         'pytest==3.4.2',

pyrasterframes/src/main/python/tests/GeoTiffWriterTests.py

@@ -37,7 +37,7 @@ def test_identity_write(self):
         dest = self._tmpfile()
         rf.write.geotiff(dest)
 
-        rf2 = self.spark.read.geotiff(dest)
+        rf2 = self.spark.read.geotiff('file://' + dest)
         self.assertEqual(rf2.count(), rf.count())
 
         os.remove(dest)
@@ -47,7 +47,7 @@ def test_unstructured_write(self):
         dest = self._tmpfile()
         rf.write.geotiff(dest, crs='EPSG:32616')
 
-        rf2 = self.spark.read.raster(dest)
+        rf2 = self.spark.read.raster('file://' + dest)
         self.assertEqual(rf2.count(), rf.count())
 
         os.remove(dest)

pyrasterframes/src/main/python/tests/PyRasterFramesTests.py

@@ -28,7 +28,6 @@
 from . import TestEnvironment
 
 
-
 class CellTypeHandling(unittest.TestCase):
 
     def test_is_raw(self):
@@ -237,11 +236,16 @@ def less_pi(t):
 class TileOps(TestEnvironment):
 
     def setUp(self):
+        from pyspark.sql import Row
         # convenience so we can assert around Tile() == Tile()
         self.t1 = Tile(np.array([[1, 2],
                                  [3, 4]]), CellType.int8().with_no_data_value(3))
         self.t2 = Tile(np.array([[1, 2],
                                  [3, 4]]), CellType.int8().with_no_data_value(1))
+        self.t3 = Tile(np.array([[1,  2],
+                                 [-3, 4]]), CellType.int8().with_no_data_value(3))
+
+        self.df = self.spark.createDataFrame([Row(t1=self.t1, t2=self.t2, t3=self.t3)])
 
     def test_addition(self):
         e1 = np.ma.masked_equal(np.array([[5, 6],
@@ -253,6 +257,9 @@ def test_addition(self):
         r2 = (self.t1 + self.t2).cells
         self.assertTrue(np.ma.allequal(r2, e2))
 
+        col_result = self.df.select(rf_local_add('t1', 't3').alias('sum')).first()
+        self.assertEqual(col_result.sum, self.t1 + self.t3)
+
     def test_multiplication(self):
         e1 = np.ma.masked_equal(np.array([[4, 8],
                                           [12, 16]]), 12)
@@ -263,6 +270,9 @@ def test_multiplication(self):
         r2 = (self.t1 * self.t2).cells
         self.assertTrue(np.ma.allequal(r2, e2))
 
+        r3 = self.df.select(rf_local_multiply('t1', 't3').alias('r3')).first().r3
+        self.assertEqual(r3, self.t1 * self.t3)
+
     def test_subtraction(self):
         t3 = self.t1 * 4
         r1 = t3 - self.t1
@@ -541,7 +551,6 @@ def path(scene, band):
         self.assertTrue(df2.select('b1_path').distinct().count() == 3)
 
 
-
 def suite():
     function_tests = unittest.TestSuite()
     return function_tests

pyrasterframes/src/main/python/tests/RasterFunctionsTests.py

@@ -97,6 +97,21 @@ def test_agg_mean(self):
         mean = self.rf.agg(rf_agg_mean('tile')).first()['rf_agg_mean(tile)']
         self.assertTrue(self.rounded_compare(mean, 10160))
 
+    def test_agg_local_mean(self):
+        from pyspark.sql import Row
+
+        # this is really testing the nodata propagation in the agg  local summation
+        ct = CellType.int8().with_no_data_value(4)
+        df = self.spark.createDataFrame([
+            Row(tile=Tile(np.array([[1, 2, 3, 4, 5, 6]]), ct)),
+            Row(tile=Tile(np.array([[1, 2, 4, 3, 5, 6]]), ct)),
+        ])
+
+        result = df.agg(rf_agg_local_mean('tile').alias('mean')).first().mean
+
+        expected = Tile(np.array([[1.0, 2.0, 3.0, 3.0, 5.0, 6.0]]), CellType.float64())
+        self.assertEqual(result, expected)
+
     def test_aggregations(self):
         aggs = self.rf.agg(
             rf_agg_data_cells('tile'),
@@ -112,28 +127,59 @@ def test_aggregations(self):
         self.assertEqual(row['rf_agg_stats(tile)'].data_cells, row['rf_agg_data_cells(tile)'])
 
     def test_sql(self):
+
         self.rf.createOrReplaceTempView("rf_test_sql")
 
-        self.spark.sql("""SELECT tile, 
-                            rf_local_add(tile, 1) AS and_one, 
-                            rf_local_subtract(tile, 1) AS less_one, 
-                            rf_local_multiply(tile, 2) AS times_two, 
-                            rf_local_divide(tile, 2) AS over_two 
-                        FROM rf_test_sql""").createOrReplaceTempView('rf_test_sql_1')
-
-        statsRow = self.spark.sql("""
-        SELECT rf_tile_mean(tile) as base,
-            rf_tile_mean(and_one) as plus_one,
-            rf_tile_mean(less_one) as minus_one,
-            rf_tile_mean(times_two) as double,
-            rf_tile_mean(over_two) as half
-        FROM rf_test_sql_1
-        """).first()
-
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.plus_one - 1))
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.minus_one + 1))
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.double / 2))
-        self.assertTrue(self.rounded_compare(statsRow.base, statsRow.half * 2))
+        arith = self.spark.sql("""SELECT tile, 
+                                rf_local_add(tile, 1) AS add_one, 
+                                rf_local_subtract(tile, 1) AS less_one, 
+                                rf_local_multiply(tile, 2) AS times_two, 
+                                rf_local_divide(
+                                    rf_convert_cell_type(tile, "float32"), 
+                                    2) AS over_two 
+                            FROM rf_test_sql""")
+
+        arith.createOrReplaceTempView('rf_test_sql_1')
+        arith.show(truncate=False)
+        stats = self.spark.sql("""
+            SELECT rf_tile_mean(tile) as base,
+                rf_tile_mean(add_one) as plus_one,
+                rf_tile_mean(less_one) as minus_one,
+                rf_tile_mean(times_two) as double,
+                rf_tile_mean(over_two) as half,
+                rf_no_data_cells(tile) as nd
+                
+            FROM rf_test_sql_1
+            ORDER BY rf_no_data_cells(tile)
+            """)
+        stats.show(truncate=False)
+        stats.createOrReplaceTempView('rf_test_sql_stats')
+
+        compare = self.spark.sql("""
+            SELECT 
+                plus_one - 1.0 = base as add,
+                minus_one + 1.0 = base as subtract,
+                double / 2.0 = base as multiply,
+                half * 2.0 = base as divide,
+                nd
+            FROM rf_test_sql_stats
+            """)
+
+        expect_row1 = compare.orderBy('nd').first()
+
+        self.assertTrue(expect_row1.subtract)
+        self.assertTrue(expect_row1.multiply)
+        self.assertTrue(expect_row1.divide)
+        self.assertEqual(expect_row1.nd, 0)
+        self.assertTrue(expect_row1.add)
+
+        expect_row2 = compare.orderBy('nd', ascending=False).first()
+
+        self.assertTrue(expect_row2.subtract)
+        self.assertTrue(expect_row2.multiply)
+        self.assertTrue(expect_row2.divide)
+        self.assertTrue(expect_row2.nd > 0)
+        self.assertTrue(expect_row2.add)  # <-- Would fail in a case where ND + 1 = 1
 
     def test_explode(self):
         import pyspark.sql.functions as F