GTC-3077 Give NoIntersectionError in AFi if geometry too small

In the AFi analysis, we currently don't return any row at all for a
location whose geometry does not intersect the centroid of any pixel
(mostly because the geometry is small compared to the pixel size). Add
similar code as in FCD to detect this case and return
NoIntersectionError.

Same change for GFWProDashboard as well.

Added a NoIntersection test for AFi.

src/main/scala/org/globalforestwatch/summarystats/afi/AFiAnalysis.scala

@@ -11,6 +11,7 @@ import org.apache.spark.storage.StorageLevel
 import org.apache.spark.sql.functions._
 import scala.collection.immutable.SortedMap
 import io.circe.syntax._
+import cats.data.Validated.{Invalid, Valid}
 
 object AFiAnalysis extends SummaryAnalysis {
 
@@ -42,13 +43,22 @@ object AFiAnalysis extends SummaryAnalysis {
 
     import spark.implicits._
 
+    // If a location has no AFiDataGroup entries, then the geometry must not have
+    // intersected the centroid of any pixels, so report the location as
+    // NoIntersectionError.
+     val summary1RDD = summaryRDD.map {
+       case Valid(Location(fid, data)) if data.isEmpty =>
+         Invalid(Location(fid, NoIntersectionError))
+       case data => data
+     }
+
     // Null out gadm_id for all non-dissolved rows and then aggregate all results for
     // each unique (list_id, location_id, gadm_id, loss_year). Need to combine first
     // with key including loss_year, so we don't have duplicate loss year entries
     // when we create the map of loss years.
     val summary1DF = AFiAnalysis.aggregateByLossYear(
         AFiDF
-        .getFeatureDataFrame(summaryRDD, spark)
+        .getFeatureDataFrame(summary1RDD, spark)
         .withColumn(
           "gadm_id", when(col("location_id") =!= -1, lit("") ).otherwise(col("gadm_id"))
         )

src/main/scala/org/globalforestwatch/summarystats/forest_change_diagnostic/ForestChangeDiagnosticAnalysis.scala

@@ -90,6 +90,9 @@ object ForestChangeDiagnosticAnalysis extends SummaryAnalysis {
           .persist(StorageLevel.MEMORY_AND_DISK)
       }
 
+    // If a location has empty ForestChangeDiagnosticData results, then the geometry
+    // must not have intersected the centroid of any pixels, so report the location
+    // as NoIntersectionError.
       partialResult.map {
         case Valid(Location(fid, data)) if data.equals(ForestChangeDiagnosticData.empty) =>
           Invalid(Location(fid, NoIntersectionError))

src/main/scala/org/globalforestwatch/summarystats/gfwpro_dashboard/GfwProDashboardAnalysis.scala

@@ -18,6 +18,7 @@ import org.globalforestwatch.util.GeotrellisGeometryValidator.makeValidGeom
 import scala.collection.JavaConverters._
 import java.time.LocalDate
 import org.globalforestwatch.util.IntersectGeometry
+import cats.data.Validated.{Invalid, Valid}
 
 object GfwProDashboardAnalysis extends SummaryAnalysis {
 
@@ -109,7 +110,18 @@ object GfwProDashboardAnalysis extends SummaryAnalysis {
           val validatedSummaryStatsRdd = GfwProDashboardRDD(tmp,
                GfwProDashboardGrid.blockTileGrid,
                kwargs + ("getRasterGadm" -> !doGadmIntersect))
-          ValidatedWorkflow(validatedSummaryStatsRdd).mapValid { summaryStatsRDD =>
+          validatedSummaryStatsRdd.collect().foreach(println)
+
+          // If a location has no GfwProDashboardRawDataGroup entries, then the
+          // geometry must not have intersected the centroid of any pixels, so report
+          // the location as NoIntersectionError.
+          val validatedSummaryStatsRdd1 = validatedSummaryStatsRdd.map {
+            case Valid(Location(fid, data)) if data.isEmpty =>
+              Invalid(Location(fid, NoIntersectionError))
+            case data => data
+          }
+
+          ValidatedWorkflow(validatedSummaryStatsRdd1).mapValid { summaryStatsRDD =>
             summaryStatsRDD
               .flatMap { case (CombinedFeatureId(fid@GfwProFeatureId(listId, locationId), gadmId), summary) =>
                 // For non-dissolved locations or vector gadm intersection, merge all

src/main/scala/org/globalforestwatch/summarystats/gfwpro_dashboard/GfwProDashboardDF.scala

@@ -2,7 +2,7 @@ package org.globalforestwatch.summarystats.gfwpro_dashboard
 
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{DataFrame, SparkSession}
-import org.globalforestwatch.features.{FeatureId, GfwProFeatureId, CombinedFeatureId, GadmFeatureId}
+import org.globalforestwatch.features.{FeatureId, GfwProFeatureId, CombinedFeatureId}
 import org.globalforestwatch.summarystats._
 import cats.data.Validated.{Valid, Invalid}
 import org.apache.spark.sql.functions.expr

src/test/scala/org/globalforestwatch/summarystats/afi/AFiAnalysisSpec.scala

@@ -11,7 +11,7 @@ import org.apache.spark.sql.{DataFrame, SaveMode}
 import org.globalforestwatch.{TestEnvironment, ProTag}
 import org.globalforestwatch.config.GfwConfig
 import org.globalforestwatch.features.{FeatureFilter, GfwProFeatureId, ValidatedFeatureRDD}
-import org.globalforestwatch.summarystats.{Location, ValidatedLocation}
+import org.globalforestwatch.summarystats.{Location, ValidatedLocation,JobError}
 
 class AFiAnalysisSpec extends TestEnvironment with DataFrameComparer {
   def palm32InputTsvPath = getClass.getResource("/palm-oil-32.tsv").toString()
@@ -54,6 +54,23 @@ class AFiAnalysisSpec extends TestEnvironment with DataFrameComparer {
       .withColumn("location_error", lit(""))
   }
 
+  it("reports no-intersection geometry as invalid") {
+    // This is an area in Paraguay (60W, 20S) that is so tiny (about 1m by 1m) that
+    // it doesn't intersect the centroid of any pixel, so it should return a row with
+    // a NoIntersectionError (rather than no row at all or a row with just empty/zero
+    // results).
+    val x = -60
+    val y = -20
+    val smallGeom = Polygon(LineString(Point(x+0,y+0), Point(x+0.00001,y+0), Point(x+0.00001,y+0.00001), Point(x+0,y+0.00001), Point(x+0,y+0)))
+    val featureRDD = spark.sparkContext.parallelize(List(
+      Validated.valid[Location[JobError], Location[Geometry]](Location(GfwProFeatureId("1", 1), smallGeom))
+    ))
+    val afi = AFI(featureRDD)
+    val res = afi.collect()
+    res.length shouldBe 1
+    res.head.getAs[String]("location_error") shouldBe "[\"NoIntersectionError\"]"
+  }
+
   it("matches recorded output for palm oil mills location 31", ProTag) {
     val featureLoc31RDD = ValidatedFeatureRDD(
       NonEmptyList.one(palm32InputTsvPath),