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.utils.stats
10 
11 import com.typesafe.scalalogging.LazyLogging
12 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
13 import org.locationtech.geomesa.curve.TimePeriod.TimePeriod
14 import org.locationtech.geomesa.curve.{BinnedTime, Z3SFC}
15 import org.locationtech.geomesa.utils.clearspring.CountMinSketch
16 import org.locationtech.jts.geom.Geometry
17 
18 import java.util.Date
19 import scala.collection.immutable.ListMap
20 
21 /**
22   * Estimates frequency counts at scale. Tracks geometry and date attributes as a single value.
23   *
24   * @param sft simple feature type
25   * @param geom geometry attribute in the sft
26   * @param dtg date attribute in the sft
27   * @param period time period to use for z index
28   * @param precision number of bits of z-index that will be used
29   * @param eps (epsilon) with probability at least @see confidence, estimates will be within eps * N
30   * @param confidence percent - with probability at least confidence, estimates will be within @see eps * N
31   */
32 class Z3Frequency(
33     val sft: SimpleFeatureType,
34     val geom: String,
35     val dtg: String,
36     val period: TimePeriod,
37     val precision: Int,
38     val eps: Double = 0.005,
39     val confidence: Double = 0.95
40   ) extends Stat with LazyLogging {
41 
42   override type S = Z3Frequency
43 
44   private val g = sft.indexOf(geom)
45   private val d = sft.indexOf(dtg)
46 
47   private val mask = Frequency.getMask(precision)
48   private val sfc = Z3SFC(period)
49   private val timeToBin = BinnedTime.timeToBinnedTime(period)
50 
51   private [stats] val sketches = scala.collection.mutable.Map.empty[Short, CountMinSketch]
52   private [stats] def newSketch: CountMinSketch = CountMinSketch(eps, confidence, Frequency.Seed)
53 
54   private def toKey(geom: Geometry, dtg: Date): (Short, Long) = {
55     import org.locationtech.geomesa.utils.geotools.Conversions.RichGeometry
56     val BinnedTime(b, o) = timeToBin(dtg.getTime)
57     val centroid = geom.safeCentroid()
58     val z = sfc.index(centroid.getX, centroid.getY, o) & mask
59     (b, z)
60   }
61 
62   /**
63     * Gets the count for the given values
64     *
65     * @param geom geometry
66     * @param dtg date
67     * @return count of the values
68     */
69   def count(geom: Geometry, dtg: Date): Long = {
70     val (bin, z3) = toKey(geom, dtg)
71     countDirect(bin, z3)
72   }
73 
74   /**
75     * Gets the count for a time bin and z3. Useful if the values are known ahead of time.
76     *
77     * @param bin period since the epoch
78     * @param z3 z value
79     * @return count of the values
80     */
81   def countDirect(bin: Short, z3: Long): Long = sketches.get(bin).map(_.estimateCount(z3)).getOrElse(0L)
82 
83   /**
84     * Number of observations in the frequency map
85     *
86     * @return number of observations
87     */
88   def size: Long = sketches.values.map(_.size).sum
89 
90   /**
91     * Split the stat into a separate stat per time bin of z data. Allows for separate handling of the reduced
92     * data set.
93     *
94     * @return
95     */
96   def splitByTime: Seq[(Short, Z3Frequency)] = {
97     sketches.toSeq.map { case (w, sketch) =>
98       val freq = new Z3Frequency(sft, geom, dtg, period, precision, eps, confidence)
99       freq.sketches.put(w, sketch)
100       (w, freq)
101     }
102   }
103 
104   override def observe(sf: SimpleFeature): Unit = {
105     val geom = sf.getAttribute(g).asInstanceOf[Geometry]
106     val dtg  = sf.getAttribute(d).asInstanceOf[Date]
107     if (geom != null && dtg != null) {
108       try {
109         val (bin, z3) = toKey(geom, dtg)
110         sketches.getOrElseUpdate(bin, newSketch).add(z3, 1L)
111       } catch {
112         case e: Exception => logger.warn(s"Error observing geom '$geom' and date '$dtg': ${e.toString}")
113       }
114     }
115   }
116 
117   // no-op
118   override def unobserve(sf: SimpleFeature): Unit = {}
119 
120   override def +(other: Z3Frequency): Z3Frequency = {
121     val plus = new Z3Frequency(sft, geom, dtg, period, precision, eps, confidence)
122     plus += this
123     plus += other
124     plus
125   }
126 
127   override def +=(other: Z3Frequency): Unit = {
128     other.sketches.filter(_._2.size > 0).foreach { case (w, sketch) =>
129       sketches.getOrElseUpdate(w, newSketch) += sketch
130     }
131   }
132 
133   override def clear(): Unit = sketches.values.foreach(_.clear())
134 
135   override def isEmpty: Boolean = sketches.values.forall(_.size == 0)
136 
137   override def toJsonObject: Any = {
138     val (e, c) = sketches.values.headOption.map(s => (s.eps, s.confidence)).getOrElse((0d, 0d))
139     ListMap("eps" -> e, "confidence" -> c, "size" -> size)
140   }
141 
142   override def isEquivalent(other: Stat): Boolean = {
143     other match {
144       case s: Z3Frequency =>
145         geom == s.geom && dtg == s.dtg && period == s.period && precision == s.precision && {
146           val nonEmpty = sketches.filter(_._2.size > 0)
147           val sNonEmpty = s.sketches.filter(_._2.size > 0)
148           nonEmpty.keys == sNonEmpty.keys && nonEmpty.keys.forall { k =>
149             sketches(k).isEquivalent(s.sketches(k))
150           }
151         }
152       case _ => false
153     }
154   }
155 }