1 /***********************************************************************
2  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * http://www.opensource.org/licenses/apache2.0.php.
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.process.query
10 
11 import com.typesafe.scalalogging.LazyLogging
12 import org.geotools.api.data.{Query, SimpleFeatureSource}
13 import org.geotools.api.feature.Feature
14 import org.geotools.api.feature.simple.SimpleFeature
15 import org.geotools.data.collection.ListFeatureCollection
16 import org.geotools.data.simple.SimpleFeatureCollection
17 import org.geotools.process.factory.{DescribeParameter, DescribeProcess, DescribeResult}
18 import org.locationtech.geomesa.filter.factory.FastFilterFactory
19 import org.locationtech.geomesa.index.geotools.GeoMesaFeatureCollection
20 import org.locationtech.geomesa.index.process.GeoMesaProcessVisitor
21 import org.locationtech.geomesa.process.{FeatureResult, GeoMesaProcess}
22 import org.locationtech.geomesa.utils.collection.SelfClosingIterator
23 import org.locationtech.geomesa.utils.geotools.Conversions._
24 
25 @DescribeProcess(
26   title = "Geomesa-enabled Proximity Search",
27   description = "Performs a proximity search on a Geomesa feature collection using another feature collection as input"
28 )
29 class ProximitySearchProcess extends GeoMesaProcess with LazyLogging {
30 
31   @DescribeResult(description = "Output feature collection")
32   def execute(
33                @DescribeParameter(
34                  name = "inputFeatures",
35                  description = "Input feature collection that defines the proximity search")
36                inputFeatures: SimpleFeatureCollection,
37 
38                @DescribeParameter(
39                  name = "dataFeatures",
40                  description = "The data set to query for matching features")
41                dataFeatures: SimpleFeatureCollection,
42 
43                @DescribeParameter(
44                  name = "bufferDistance",
45                  description = "Buffer size in meters")
46                bufferDistance: java.lang.Double
47 
48                ): SimpleFeatureCollection = {
49 
50     logger.debug(s"Attempting Geomesa Proximity Search on collection type ${dataFeatures.getClass.getName}")
51 
52     val visitor = new ProximityVisitor(inputFeatures, dataFeatures, bufferDistance.doubleValue())
53     GeoMesaFeatureCollection.visit(dataFeatures, visitor)
54     visitor.getResult.results
55   }
56 }
57 
58 class ProximityVisitor(inputFeatures: SimpleFeatureCollection,
59                        dataFeatures: SimpleFeatureCollection,
60                        bufferInMeters: Double) extends GeoMesaProcessVisitor with LazyLogging {
61 
62   import org.locationtech.geomesa.filter.{ff, mergeFilters, orFilters}
63 
64   private val dwithin = {
65     val geomProperty = ff.property(dataFeatures.getSchema.getGeometryDescriptor.getName)
66     val geomFilters = SelfClosingIterator(inputFeatures.features).map { sf =>
67       ff.dwithin(geomProperty, ff.literal(sf.geometry), bufferInMeters, "meters")
68     }
69     orFilters(geomFilters.toSeq)
70   }
71 
72   // normally handled in our query planner, but we are going to use the filter directly here
73   private lazy val manualFilter = FastFilterFactory.optimize(dataFeatures.getSchema, dwithin)
74   private val manualVisitResults = new ListFeatureCollection(dataFeatures.getSchema)
75 
76   private var resultCalc = FeatureResult(manualVisitResults)
77 
78   // non-optimized visit
79   // here we use degrees for our filters since we are manually evaluating them.
80   override def visit(feature: Feature): Unit = {
81     val sf = feature.asInstanceOf[SimpleFeature]
82     if (manualFilter.evaluate(sf)) {
83       manualVisitResults.add(sf)
84     }
85   }
86 
87   override def getResult: FeatureResult = resultCalc
88 
89   override def execute(source: SimpleFeatureSource, query: Query): Unit = {
90     logger.debug(s"Running Geomesa Proximity Search on source type ${source.getClass.getName}")
91     val combinedFilter = mergeFilters(query.getFilter, dwithin)
92     resultCalc = FeatureResult(source.getFeatures(combinedFilter))
93   }
94 }