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.analytic
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.features.ScalaSimpleFeature
19 import org.locationtech.geomesa.index.geotools.GeoMesaFeatureCollection
20 import org.locationtech.geomesa.index.iterators.StatsScan
21 import org.locationtech.geomesa.index.process.GeoMesaProcessVisitor
22 import org.locationtech.geomesa.index.stats.HasGeoMesaStats
23 import org.locationtech.geomesa.process.analytic.MinMaxProcess.MinMaxVisitor
24 import org.locationtech.geomesa.process.{FeatureResult, GeoMesaProcess}
25 import org.locationtech.geomesa.utils.collection.SelfClosingIterator
26 import org.locationtech.geomesa.utils.geotools.GeometryUtils
27 import org.locationtech.geomesa.utils.stats.Stat
28 
29 @DescribeProcess(
30   title = "Min/Max Process",
31   description = "Gets attribute bounds for a data set"
32 )
33 class MinMaxProcess extends GeoMesaProcess with LazyLogging {
34 
35   @DescribeResult(description = "Output feature collection")
36   def execute(
37                  @DescribeParameter(
38                    name = "features",
39                    description = "The feature set on which to query")
40                  features: SimpleFeatureCollection,
41 
42                  @DescribeParameter(
43                    name = "attribute",
44                    description = "The attribute to gather bounds for")
45                  attribute: String,
46 
47                  @DescribeParameter(
48                    name = "cached",
49                    description = "Return cached values, if available",
50                    min = 0, max = 1)
51                  cached: java.lang.Boolean = null
52 
53              ): SimpleFeatureCollection = {
54 
55     require(attribute != null, "Attribute is a required field")
56 
57     logger.debug(s"Attempting min/max process on type ${features.getClass.getName}")
58 
59     val visitor = new MinMaxVisitor(features, attribute, Option(cached).forall(_.booleanValue()))
60     GeoMesaFeatureCollection.visit(features, visitor)
61     visitor.getResult.results
62   }
63 }
64 
65 object MinMaxProcess {
66 
67   class MinMaxVisitor(features: SimpleFeatureCollection, attribute: String, cached: Boolean)
68       extends GeoMesaProcessVisitor with LazyLogging {
69 
70     private lazy val stat: Stat = Stat(features.getSchema, Stat.MinMax(attribute))
71 
72     private var resultCalc: FeatureResult = _
73 
74     // non-optimized visit
75     override def visit(feature: Feature): Unit = stat.observe(feature.asInstanceOf[SimpleFeature])
76 
77     override def getResult: FeatureResult = {
78       if (resultCalc != null) {
79         resultCalc
80       } else {
81         createResult(stat.toJson)
82       }
83     }
84 
85     override def execute(source: SimpleFeatureSource, query: Query): Unit = {
86       logger.debug(s"Running Geomesa min/max process on source type ${source.getClass.getName}")
87 
88       source.getDataStore match {
89         case ds: HasGeoMesaStats =>
90           resultCalc = ds.stats.getMinMax[Any](source.getSchema, attribute, query.getFilter, !cached) match {
91             case None     => createResult("{}")
92             case Some(mm) => createResult(mm.toJson)
93           }
94 
95         case ds =>
96           logger.warn(s"Running unoptimized min/max query on ${ds.getClass.getName}")
97           SelfClosingIterator(features.features).foreach(visit)
98       }
99     }
100   }
101 
102   private def createResult(stat: String): FeatureResult = {
103     val sf = new ScalaSimpleFeature(StatsScan.StatsSft, "", Array(stat, GeometryUtils.zeroPoint))
104     FeatureResult(new ListFeatureCollection(StatsScan.StatsSft, sf))
105   }
106 }