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.geotools.geometry.jts.JTSFactoryFinder
14 import org.locationtech.geomesa.curve.TimePeriod.TimePeriod
15 import org.locationtech.geomesa.curve.{BinnedTime, TimePeriod, Z3SFC}
16 import org.locationtech.geomesa.utils.stats.BinnedArray.LongBinning
17 import org.locationtech.geomesa.utils.stats.MinMax.MinMaxGeometry
18 import org.locationtech.jts.geom.{Coordinate, Geometry, Point}
19 
20 import java.util.Date
21 
22 /**
23   * The histogram's state is stored in an indexed array, where the index is the bin number
24   * and the values are the counts.
25   *
26   * Tracks geometry and date attributes as a single value.
27   *
28   * @param sft simple feature type
29   * @param geom geometry attribute in the sft
30   * @param dtg date attribute in the sft
31   * @param period time period to use for z index
32   * @param length number of bins the histogram has, per period
33  */
34 class Z3Histogram(
35     val sft: SimpleFeatureType,
36     val geom: String,
37     val dtg: String,
38     val period: TimePeriod,
39     val length: Int
40   ) extends Stat with LazyLogging {
41 
42   import Z3Histogram._
43 
44   override type S = Z3Histogram
45 
46   private val g = sft.indexOf(geom)
47   private val d = sft.indexOf(dtg)
48 
49   private val sfc = Z3SFC(period)
50   private val timeToBin = BinnedTime.timeToBinnedTime(period)
51   private val binToDate = BinnedTime.binnedTimeToDate(period)
52   private val minZ = sfc.index(minGeom.getX, minGeom.getY, sfc.time.min.toLong)
53   private val maxZ = sfc.index(maxGeom.getX, maxGeom.getY, sfc.time.max.toLong)
54 
55   private lazy val jsonFormat = period match {
56     case TimePeriod.Day   => s"$period-%05d"
57     case TimePeriod.Week  => s"$period-%04d"
58     case TimePeriod.Month => s"$period-%03d"
59     case TimePeriod.Year  => s"$period-%02d"
60   }
61 
62   private [stats] val binMap = scala.collection.mutable.Map.empty[Short, BinnedArray[java.lang.Long]]
63   private [stats] def newBins = new BinnedArray(new LongBinning(length, (minZ, maxZ)))
64 
65   def timeBins: Seq[Short] = binMap.keys.toSeq.sorted
66   def count(timeBin: Short, i: Int): Long = binMap.get(timeBin).map(_.counts(i)).getOrElse(0L)
67 
68   def directIndex(timeBin: Short, value: Long): Int = binMap.get(timeBin).map(_.indexOf(value)).getOrElse(-1)
69 
70   def indexOf(value: (Geometry, Date)): (Short, Int) = {
71     val (timeBin, z) = toKey(value._1, value._2, lenient = false)
72     (timeBin, directIndex(timeBin, z))
73   }
74 
75   def medianValue(timeBin: Short, i: Int): (Geometry, Date) = fromKey(timeBin, binMap(timeBin).medianValue(i))
76 
77   private def toKey(geom: Geometry, dtg: Date, lenient: Boolean): (Short, Long) = {
78     import org.locationtech.geomesa.utils.geotools.Conversions.RichGeometry
79     val BinnedTime(bin, offset) = timeToBin(dtg.getTime)
80     val centroid = geom.safeCentroid()
81     val z = sfc.index(centroid.getX, centroid.getY, offset, lenient)
82     (bin, z)
83   }
84 
85   private def fromKey(timeBin: Short, z: Long): (Geometry, Date) = {
86     val (x, y, t) = sfc.invert(z)
87     val dtg = Date.from(binToDate(BinnedTime(timeBin, t)).toInstant)
88     val geom = Z3Histogram.gf.createPoint(new Coordinate(x, y))
89     (geom, dtg)
90   }
91 
92   /**
93     * Split the stat into a separate stat per time bin of z data. Allows for separate handling of the reduced
94     * data set.
95     *
96     * @return
97     */
98   def splitByTime: Seq[(Short, Z3Histogram)] = {
99     binMap.toSeq.map { case (w, bins) =>
100       val hist = new Z3Histogram(sft, geom, dtg, period, length)
101       hist.binMap.put(w, bins)
102       (w, hist)
103     }
104   }
105 
106   override def observe(sf: SimpleFeature): Unit = {
107     val geom = sf.getAttribute(g).asInstanceOf[Geometry]
108     val dtg  = sf.getAttribute(d).asInstanceOf[Date]
109     if (geom != null && dtg != null) {
110       try {
111         val (timeBin, z3) = toKey(geom, dtg, lenient = false)
112         binMap.getOrElseUpdate(timeBin, newBins).add(z3, 1L)
113       } catch {
114         case e: Exception => logger.warn(s"Error observing geom '$geom' and date '$dtg': ${e.toString}")
115       }
116     }
117   }
118 
119   override def unobserve(sf: SimpleFeature): Unit = {
120     val geom = sf.getAttribute(g).asInstanceOf[Geometry]
121     val dtg  = sf.getAttribute(d).asInstanceOf[Date]
122     if (geom != null && dtg != null) {
123       try {
124         val (timeBin, z3) = toKey(geom, dtg, lenient = true)
125         binMap.get(timeBin).foreach(_.add(z3, -1L))
126       } catch {
127         case e: Exception => logger.warn(s"Error un-observing geom '$geom' and date '$dtg': ${e.toString}")
128       }
129     }
130   }
131 
132   /**
133     * Creates a new histogram by combining another histogram with this one
134     */
135   override def +(other: Z3Histogram): Z3Histogram = {
136     val plus = new Z3Histogram(sft, geom, dtg, period, length)
137     plus += this
138     plus += other
139     plus
140   }
141 
142   /**
143     * Copies another histogram into this one
144     */
145   override def +=(other: Z3Histogram): Unit = {
146     if (length != other.length) {
147       throw new NotImplementedError("Can only add z3 histograms with the same length")
148     }
149     other.binMap.foreach { case (w, bins) =>
150       binMap.get(w) match {
151         case None => binMap.put(w, bins) // note: sharing a reference now
152         case Some(b) =>
153           var i = 0
154           while (i < b.length) {
155             b.counts(i) += bins.counts(i)
156             i += 1
157           }
158       }
159     }
160   }
161 
162   override def toJsonObject: Any =
163     binMap.toSeq.sortBy(_._1)
164       .map { case (p, bins) => (String.format(jsonFormat, Short.box(p)), bins) }
165       .map { case (label, bins) => Map(label-> Map("bins" -> bins.counts)) }
166 
167   override def isEmpty: Boolean = binMap.values.forall(_.counts.forall(_ == 0))
168 
169   override def clear(): Unit = binMap.values.foreach(_.clear())
170 
171   override def isEquivalent(other: Stat): Boolean = other match {
172     case that: Z3Histogram =>
173       g == that.g && d == that.d && period == that.period &&
174           length == that.length && binMap.keySet == that.binMap.keySet &&
175           binMap.forall { case (w, bins) => java.util.Arrays.equals(bins.counts, that.binMap(w).counts) }
176     case _ => false
177   }
178 }
179 
180 object Z3Histogram {
181 
182   private val gf = JTSFactoryFinder.getGeometryFactory
183 
184   val minGeom: Point = MinMaxGeometry.min.asInstanceOf[Point]
185   val maxGeom: Point = MinMaxGeometry.max.asInstanceOf[Point]
186 }