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.transform
10 
11 import org.geotools.data.collection.ListFeatureCollection
12 import org.geotools.data.simple.SimpleFeatureCollection
13 import org.geotools.feature.simple.{SimpleFeatureBuilder, SimpleFeatureTypeBuilder}
14 import org.geotools.process.ProcessException
15 import org.geotools.process.factory.{DescribeParameter, DescribeProcess, DescribeResult}
16 import org.locationtech.geomesa.process.GeoMesaProcess
17 import org.locationtech.geomesa.utils.collection.SelfClosingIterator
18 
19 import java.nio.charset.StandardCharsets
20 import scala.util.hashing.MurmurHash3
21 
22 trait HashAttribute {
23 
24   def transformHash(hash: Int): AnyRef
25   // note - augmentSft needs to add an attribute called 'hash'
26   def augmentSft(sft: SimpleFeatureTypeBuilder): Unit
27 
28   @throws(classOf[ProcessException])
29   @DescribeResult(name = "result", description = "Output collection")
30   def execute(@DescribeParameter(name = "data", description = "Input features")
31               obsFeatures: SimpleFeatureCollection,
32               @DescribeParameter(name = "attribute", description = "The attribute to hash on")
33               attribute: String,
34               @DescribeParameter(name = "modulo", description = "The divisor")
35               modulo: Integer): SimpleFeatureCollection = {
36 
37     val sft = obsFeatures.getSchema
38     val sftBuilder = new SimpleFeatureTypeBuilder()
39     sftBuilder.init(sft)
40     augmentSft(sftBuilder)
41     val targetSft = sftBuilder.buildFeatureType()
42     val hashIndex = targetSft.indexOf("hash")
43     val featureBuilder = new SimpleFeatureBuilder(targetSft)
44 
45     val results = new ListFeatureCollection(targetSft)
46 
47     SelfClosingIterator(obsFeatures.features()).foreach { sf =>
48       featureBuilder.reset()
49       featureBuilder.init(sf)
50       val attr = Option(sf.getAttribute(attribute)).map(_.toString).getOrElse("")
51       val hash = math.abs(MurmurHash3.bytesHash(attr.getBytes(StandardCharsets.UTF_16LE))) % modulo
52       featureBuilder.set(hashIndex, transformHash(hash))
53       results.add(featureBuilder.buildFeature(sf.getID))
54     }
55 
56     results
57   }
58 }
59 
60 @DescribeProcess(
61   title = "Hash Attribute Process",
62   description = "Adds an attribute to each SimpleFeature that hashes the configured attribute modulo the configured param"
63 )
64 class HashAttributeProcess extends GeoMesaProcess with HashAttribute {
65   override def transformHash(hash: Int): AnyRef = Int.box(hash)
66 
67   override def augmentSft(sftBuilder: SimpleFeatureTypeBuilder): Unit = {
68     sftBuilder.add("hash", classOf[Integer])
69   }
70 }
71 
72 @DescribeProcess(
73   title = "Hash Attribute Color Process",
74   description = "Adds an attribute to each SimpleFeature that hashes the configured attribute modulo the configured param and emits a color"
75 )
76 class HashAttributeColorProcess extends GeoMesaProcess with HashAttribute {
77   val colors =
78     Array[String](
79       "#6495ED",
80       "#B0C4DE",
81       "#00FFFF",
82       "#9ACD32",
83       "#00FA9A",
84       "#FFF8DC",
85       "#F5DEB3")
86 
87   override def transformHash(hash: Int): AnyRef = colors(hash % colors.length)
88 
89   override def augmentSft(sftBuilder: SimpleFeatureTypeBuilder): Unit = {
90     sftBuilder.add("hash", classOf[String])
91   }
92 }