|
| 1 | +/* |
| 2 | + * Copyright 2014 Commonwealth Computer Research, Inc. |
| 3 | + * |
| 4 | + * Licensed under the Apache License, Version 2.0 (the License); |
| 5 | + * you may not use this file except in compliance with the License. |
| 6 | + * You may obtain a copy of the License at |
| 7 | + * |
| 8 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | + * |
| 10 | + * Unless required by applicable law or agreed to in writing, software |
| 11 | + * distributed under the License is distributed on an AS IS BASIS, |
| 12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | + * See the License for the specific language governing permissions and |
| 14 | + * limitations under the License. |
| 15 | + */ |
| 16 | + |
| 17 | + |
| 18 | +package org.locationtech.geomesa.core.iterators |
| 19 | + |
| 20 | +import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, DataOutputStream} |
| 21 | +import java.util.Date |
| 22 | +import java.{util => ju} |
| 23 | + |
| 24 | +import com.typesafe.scalalogging.slf4j.Logging |
| 25 | +import com.vividsolutions.jts.geom._ |
| 26 | +import org.apache.accumulo.core.client.IteratorSetting |
| 27 | +import org.apache.accumulo.core.data.{ByteSequence, Key, Value, Range => ARange} |
| 28 | +import org.apache.accumulo.core.iterators.{IteratorEnvironment, SortedKeyValueIterator} |
| 29 | +import org.apache.commons.codec.binary.Base64 |
| 30 | +import org.geotools.feature.simple.SimpleFeatureBuilder |
| 31 | +import org.geotools.geometry.jts.JTSFactoryFinder |
| 32 | +import org.joda.time.{DateTime, Interval} |
| 33 | +import org.locationtech.geomesa.core._ |
| 34 | +import org.locationtech.geomesa.core.data.{FeatureEncoding, SimpleFeatureDecoder, SimpleFeatureEncoder} |
| 35 | +import org.locationtech.geomesa.core.index.{IndexEntryDecoder, _} |
| 36 | +import org.locationtech.geomesa.feature.AvroSimpleFeatureFactory |
| 37 | +import org.locationtech.geomesa.utils.geotools.{SimpleFeatureTypes, TimeSnap} |
| 38 | +import org.opengis.feature.simple.SimpleFeatureType |
| 39 | + |
| 40 | +import scala.util.Random |
| 41 | + |
| 42 | +class TemporalDensityIterator(other: TemporalDensityIterator, env: IteratorEnvironment) extends SortedKeyValueIterator[Key, Value] { |
| 43 | + |
| 44 | + import org.locationtech.geomesa.core.iterators.TemporalDensityIterator.{TEMPORAL_DENSITY_FEATURE_STRING, TimeSeries} |
| 45 | + |
| 46 | + var curRange: ARange = null |
| 47 | + var result: TimeSeries = new collection.mutable.HashMap[DateTime, Long]() |
| 48 | + var projectedSFT: SimpleFeatureType = null |
| 49 | + var featureBuilder: SimpleFeatureBuilder = null |
| 50 | + var snap: TimeSnap = null |
| 51 | + var topTemporalDensityKey: Option[Key] = None |
| 52 | + var topTemporalDensityValue: Option[Value] = None |
| 53 | + protected var decoder: IndexEntryDecoder = null |
| 54 | + |
| 55 | + var simpleFeatureType: SimpleFeatureType = null |
| 56 | + var source: SortedKeyValueIterator[Key,Value] = null |
| 57 | + |
| 58 | + var topSourceKey: Key = null |
| 59 | + var topSourceValue: Value = null |
| 60 | + var originalDecoder: SimpleFeatureDecoder = null |
| 61 | + var temporalDensityFeatureEncoder: SimpleFeatureEncoder = null |
| 62 | + |
| 63 | + var dateTimeFieldName: String = null |
| 64 | + |
| 65 | + def this() = this(null, null) |
| 66 | + |
| 67 | + def init(source: SortedKeyValueIterator[Key, Value], |
| 68 | + options: ju.Map[String, String], |
| 69 | + env: IteratorEnvironment): Unit = { |
| 70 | + this.source = source |
| 71 | + |
| 72 | + val simpleFeatureTypeSpec = options.get(GEOMESA_ITERATORS_SIMPLE_FEATURE_TYPE) |
| 73 | + simpleFeatureType = SimpleFeatureTypes.createType(this.getClass.getCanonicalName, simpleFeatureTypeSpec) |
| 74 | + simpleFeatureType.decodeUserData(options, GEOMESA_ITERATORS_SIMPLE_FEATURE_TYPE) |
| 75 | + |
| 76 | + dateTimeFieldName = getDtgFieldName(simpleFeatureType).getOrElse ( throw new IllegalArgumentException("dtg field required")) |
| 77 | + |
| 78 | + // default to text if not found for backwards compatibility |
| 79 | + val encodingOpt = Option(options.get(FEATURE_ENCODING)).getOrElse(FeatureEncoding.TEXT.toString) |
| 80 | + originalDecoder = SimpleFeatureDecoder(simpleFeatureType, encodingOpt) |
| 81 | + |
| 82 | + projectedSFT = SimpleFeatureTypes.createType(simpleFeatureType.getTypeName, TEMPORAL_DENSITY_FEATURE_STRING) |
| 83 | + |
| 84 | + temporalDensityFeatureEncoder = SimpleFeatureEncoder(projectedSFT, encodingOpt) |
| 85 | + featureBuilder = AvroSimpleFeatureFactory.featureBuilder(projectedSFT) |
| 86 | + |
| 87 | + val buckets = TemporalDensityIterator.getBuckets(options) |
| 88 | + val bounds = TemporalDensityIterator.getTimeBounds(options) |
| 89 | + snap = new TimeSnap(bounds, buckets) |
| 90 | + |
| 91 | + } |
| 92 | + |
| 93 | + /** |
| 94 | + * Combines the results from the underlying iterator stack |
| 95 | + * into a single feature |
| 96 | + */ |
| 97 | + def findTop() = { |
| 98 | + // reset our 'top' (current) variables |
| 99 | + result.clear() |
| 100 | + topSourceKey = null |
| 101 | + topSourceValue = null |
| 102 | + |
| 103 | + while(source.hasTop && !curRange.afterEndKey(source.getTopKey)) { |
| 104 | + topSourceKey = source.getTopKey |
| 105 | + topSourceValue = source.getTopValue //SimpleFeature |
| 106 | + |
| 107 | + val date = originalDecoder.decode(topSourceValue).getAttribute(dateTimeFieldName).asInstanceOf[Date] |
| 108 | + val dateTime = new DateTime(date.getTime) |
| 109 | + addResultDate(dateTime) |
| 110 | + |
| 111 | + source.next() |
| 112 | + } |
| 113 | + |
| 114 | + if(topSourceKey != null) { |
| 115 | + featureBuilder.reset() |
| 116 | + featureBuilder.add(TemporalDensityIterator.encodeTimeSeries(result)) |
| 117 | + featureBuilder.add(TemporalDensityIterator.zeroPoint) //Filler value as Feature requires a geometry |
| 118 | + val feature = featureBuilder.buildFeature(Random.nextString(6)) |
| 119 | + topTemporalDensityKey = Some(topSourceKey) |
| 120 | + topTemporalDensityValue = Some(new Value(temporalDensityFeatureEncoder.encode(feature))) |
| 121 | + } |
| 122 | + } |
| 123 | + |
| 124 | + /** take a given Coordinate and add 1 to the result time that it corresponds to via the snap time */ |
| 125 | + def addResultDate(date: DateTime) = { |
| 126 | + val t: DateTime = snap.t(snap.i(date)) |
| 127 | + val cur: Long = result.get(t).getOrElse(0L) |
| 128 | + result.put(t, cur + 1L) |
| 129 | + } |
| 130 | + |
| 131 | + override def seek(range: ARange, |
| 132 | + columnFamilies: ju.Collection[ByteSequence], |
| 133 | + inclusive: Boolean): Unit = { |
| 134 | + curRange = range |
| 135 | + source.seek(range, columnFamilies, inclusive) |
| 136 | + findTop() |
| 137 | + } |
| 138 | + |
| 139 | + def hasTop: Boolean = topTemporalDensityKey.nonEmpty |
| 140 | + |
| 141 | + def getTopKey: Key = topTemporalDensityKey.orNull |
| 142 | + |
| 143 | + def getTopValue = topTemporalDensityValue.orNull |
| 144 | + |
| 145 | + def deepCopy(env: IteratorEnvironment): SortedKeyValueIterator[Key, Value] = new TemporalDensityIterator(this, env) |
| 146 | + |
| 147 | + def next(): Unit = if(!source.hasTop) { |
| 148 | + topTemporalDensityKey = None |
| 149 | + topTemporalDensityValue = None |
| 150 | + } else { |
| 151 | + findTop() |
| 152 | + } |
| 153 | +} |
| 154 | + |
| 155 | +object TemporalDensityIterator extends Logging { |
| 156 | + |
| 157 | + val INTERVAL_KEY = "geomesa.temporal.density.bounds" |
| 158 | + val BUCKETS_KEY = "geomesa.temporal.density.buckets" |
| 159 | + val ENCODED_TIME_SERIES: String = "timeseries" |
| 160 | + val TEMPORAL_DENSITY_FEATURE_STRING = s"$ENCODED_TIME_SERIES:String,geom:Geometry" |
| 161 | + |
| 162 | + val zeroPoint = new GeometryFactory().createPoint(new Coordinate(0,0)) |
| 163 | + |
| 164 | + type TimeSeries = collection.mutable.HashMap[DateTime, Long] |
| 165 | + |
| 166 | + val geomFactory = JTSFactoryFinder.getGeometryFactory |
| 167 | + |
| 168 | + def configure(cfg: IteratorSetting, interval : Interval, buckets: Int) = { |
| 169 | + setTimeBounds(cfg, interval) |
| 170 | + setBuckets(cfg, buckets) |
| 171 | + } |
| 172 | + |
| 173 | + def setTimeBounds(iterSettings: IteratorSetting, interval: Interval) : Unit = { |
| 174 | + iterSettings.addOption(INTERVAL_KEY, s"${interval.getStart.getMillis},${interval.getEnd.getMillis}") |
| 175 | + } |
| 176 | + |
| 177 | + def setBuckets(iterSettings: IteratorSetting, buckets: Int): Unit = { |
| 178 | + iterSettings.addOption(BUCKETS_KEY, s"$buckets") |
| 179 | + } |
| 180 | + |
| 181 | + def getBuckets(options: ju.Map[String, String]): Int = { |
| 182 | + options.get(BUCKETS_KEY).toInt |
| 183 | + } |
| 184 | + |
| 185 | + def getTimeBounds(options: ju.Map[String, String]): Interval = { |
| 186 | + val Array(s, e) = options.get(INTERVAL_KEY).split(",").map(_.toLong) |
| 187 | + new Interval(s, e) |
| 188 | + } |
| 189 | + |
| 190 | + def combineTimeSeries(ts1: TimeSeries, ts2: TimeSeries) : TimeSeries = { |
| 191 | + val resultTS = new collection.mutable.HashMap[DateTime, Long]() |
| 192 | + for (key <- ts1.keySet ++ ts2.keySet) { |
| 193 | + resultTS.put(key, ts1.getOrElse(key, 0L) + ts2.getOrElse(key,0L)) |
| 194 | + } |
| 195 | + resultTS |
| 196 | + } |
| 197 | + |
| 198 | + def encodeTimeSeries(timeSeries: TimeSeries): String = { |
| 199 | + val baos = new ByteArrayOutputStream() |
| 200 | + val os = new DataOutputStream(baos) |
| 201 | + for((date,count) <- timeSeries) { |
| 202 | + os.writeLong(date.getMillis) |
| 203 | + os.writeLong(count) |
| 204 | + } |
| 205 | + os.flush() |
| 206 | + Base64.encodeBase64URLSafeString(baos.toByteArray) |
| 207 | + } |
| 208 | + |
| 209 | + def decodeTimeSeries(encoded: String): TimeSeries = { |
| 210 | + val bytes = Base64.decodeBase64(encoded) |
| 211 | + val is = new DataInputStream(new ByteArrayInputStream(bytes)) |
| 212 | + val table = new collection.mutable.HashMap[DateTime, Long]() |
| 213 | + while(is.available() > 0) { |
| 214 | + val dateIdx = new DateTime(is.readLong()) |
| 215 | + val weight = is.readLong() |
| 216 | + table.put(dateIdx, weight) |
| 217 | + } |
| 218 | + table |
| 219 | + } |
| 220 | +} |
0 commit comments