-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #105 from usdot-jpo-ode/tim-dedup-update
Tim dedup update
- Loading branch information
Showing
29 changed files
with
957 additions
and
359 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
20 changes: 0 additions & 20 deletions
20
jpo-deduplicator/src/main/java/us/dot/its/jpo/deduplicator/deduplicator/models/TimPair.java
This file was deleted.
Oops, something went wrong.
69 changes: 69 additions & 0 deletions
69
...main/java/us/dot/its/jpo/deduplicator/deduplicator/processors/DeduplicationProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| package us.dot.its.jpo.deduplicator.deduplicator.processors; | ||
|
|
||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
|
|
||
| import org.apache.kafka.streams.processor.PunctuationType; | ||
| import org.apache.kafka.streams.processor.api.Processor; | ||
| import org.apache.kafka.streams.processor.api.ProcessorContext; | ||
| import org.apache.kafka.streams.processor.api.Record; | ||
| import org.apache.kafka.streams.state.KeyValueStore; | ||
|
|
||
| import org.apache.kafka.streams.state.KeyValueIterator; | ||
| import org.apache.kafka.streams.KeyValue; | ||
|
|
||
| public abstract class DeduplicationProcessor<T> implements Processor<String, T, String, T>{ | ||
|
|
||
| private ProcessorContext<String, T> context; | ||
| private KeyValueStore<String, T> store; | ||
| public String storeName; | ||
|
|
||
| @Override | ||
| public void init(ProcessorContext<String, T> context) { | ||
| this.context = context; | ||
| store = context.getStateStore(storeName); | ||
| this.context.schedule(Duration.ofHours(1), PunctuationType.WALL_CLOCK_TIME, this::cleanupOldKeys); | ||
| } | ||
|
|
||
| @Override | ||
| public void process(Record<String, T> record) { | ||
|
|
||
| // Don't do anything if key is bad | ||
| if(record.key().equals("")){ | ||
| return; | ||
| } | ||
|
|
||
| T lastRecord = store.get(record.key()); | ||
| if(lastRecord == null){ | ||
| store.put(record.key(), record.value()); | ||
| context.forward(record); | ||
| return; | ||
| } | ||
|
|
||
| if(!isDuplicate(lastRecord, record.value())){ | ||
| store.put(record.key(), record.value()); | ||
| context.forward(record); | ||
| return; | ||
| } | ||
| } | ||
|
|
||
| private void cleanupOldKeys(final long timestamp) { | ||
| try (KeyValueIterator<String, T> iterator = store.all()) { | ||
| while (iterator.hasNext()) { | ||
|
|
||
| KeyValue<String, T> record = iterator.next(); | ||
| // Delete any record more than 2 hours old. | ||
| if(Instant.ofEpochMilli(timestamp).minusSeconds(2 * 60 * 60).isAfter(getMessageTime(record.value))){ | ||
| store.delete(record.key); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // returns an instant representing the time of the message | ||
| public abstract Instant getMessageTime(T message); | ||
|
|
||
| // returns if two messages are duplicates of one another | ||
| public abstract boolean isDuplicate(T lastMessage, T newMessage); | ||
|
|
||
| } |
80 changes: 80 additions & 0 deletions
80
...rc/main/java/us/dot/its/jpo/deduplicator/deduplicator/processors/OdeBsmJsonProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| package us.dot.its.jpo.deduplicator.deduplicator.processors; | ||
|
|
||
|
|
||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
| import java.time.format.DateTimeFormatter; | ||
|
|
||
| import org.geotools.referencing.GeodeticCalculator; | ||
|
|
||
| import us.dot.its.jpo.deduplicator.DeduplicatorProperties; | ||
| import us.dot.its.jpo.ode.model.OdeBsmData; | ||
| import us.dot.its.jpo.ode.model.OdeBsmMetadata; | ||
| import us.dot.its.jpo.ode.plugin.j2735.J2735Bsm; | ||
| import us.dot.its.jpo.ode.plugin.j2735.J2735BsmCoreData; | ||
|
|
||
| public class OdeBsmJsonProcessor extends DeduplicationProcessor<OdeBsmData>{ | ||
|
|
||
| DateTimeFormatter formatter = DateTimeFormatter.ISO_INSTANT; | ||
| DeduplicatorProperties props; | ||
| GeodeticCalculator calculator; | ||
|
|
||
| public OdeBsmJsonProcessor(String storeName, DeduplicatorProperties props){ | ||
| this.storeName = storeName; | ||
| this.props = props; | ||
| calculator = new GeodeticCalculator(); | ||
| } | ||
|
|
||
|
|
||
| @Override | ||
| public Instant getMessageTime(OdeBsmData message) { | ||
| try { | ||
| String time = ((OdeBsmMetadata)message.getMetadata()).getOdeReceivedAt(); | ||
| return Instant.from(formatter.parse(time)); | ||
| } catch (Exception e) { | ||
| System.out.println("Failed to Parse Time"); | ||
| return Instant.ofEpochMilli(0); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isDuplicate(OdeBsmData lastMessage, OdeBsmData newMessage) { | ||
| Instant newValueTime = getMessageTime(newMessage); | ||
| Instant oldValueTime = getMessageTime(lastMessage); | ||
|
|
||
| // If the messages are more than a certain time apart, forward the new message on | ||
| if(newValueTime.minus(Duration.ofMillis(props.getOdeBsmMaximumTimeDelta())).isAfter(oldValueTime)){ | ||
| return false; | ||
| } | ||
|
|
||
| J2735BsmCoreData oldCore = ((J2735Bsm)lastMessage.getPayload().getData()).getCoreData(); | ||
| J2735BsmCoreData newCore = ((J2735Bsm)newMessage.getPayload().getData()).getCoreData(); | ||
|
|
||
|
|
||
| // If the Vehicle is moving, forward the message on | ||
| if(newCore.getSpeed().doubleValue() > props.getOdeBsmAlwaysIncludeAtSpeed()){ | ||
| return false; | ||
| } | ||
|
|
||
|
|
||
| double distance = calculateGeodeticDistance( | ||
| newCore.getPosition().getLatitude().doubleValue(), | ||
| newCore.getPosition().getLongitude().doubleValue(), | ||
| oldCore.getPosition().getLatitude().doubleValue(), | ||
| oldCore.getPosition().getLongitude().doubleValue() | ||
| ); | ||
|
|
||
| // If the position delta between the messages is suitable large, forward the message on | ||
| if(distance > props.getOdeBsmMaximumPositionDelta()){ | ||
| return false; | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
| public double calculateGeodeticDistance(double lat1, double lon1, double lat2, double lon2) { | ||
| calculator.setStartingGeographicPoint(lon1, lat1); | ||
| calculator.setDestinationGeographicPoint(lon2, lat2); | ||
| return calculator.getOrthodromicDistance(); | ||
| } | ||
| } |
70 changes: 70 additions & 0 deletions
70
...rc/main/java/us/dot/its/jpo/deduplicator/deduplicator/processors/OdeMapJsonProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| package us.dot.its.jpo.deduplicator.deduplicator.processors; | ||
|
|
||
| import java.time.Duration; | ||
| import java.time.Instant; | ||
| import java.time.format.DateTimeFormatter; | ||
| import java.util.Objects; | ||
|
|
||
| import us.dot.its.jpo.deduplicator.DeduplicatorProperties; | ||
| import us.dot.its.jpo.ode.model.OdeMapData; | ||
| import us.dot.its.jpo.ode.model.OdeMapMetadata; | ||
| import us.dot.its.jpo.ode.model.OdeMapPayload; | ||
|
|
||
| public class OdeMapJsonProcessor extends DeduplicationProcessor<OdeMapData>{ | ||
|
|
||
| DateTimeFormatter formatter = DateTimeFormatter.ISO_INSTANT; | ||
|
|
||
| DeduplicatorProperties props; | ||
|
|
||
| public OdeMapJsonProcessor(DeduplicatorProperties props){ | ||
| this.props = props; | ||
| this.storeName = props.getKafkaStateStoreOdeMapJsonName(); | ||
| } | ||
|
|
||
|
|
||
| @Override | ||
| public Instant getMessageTime(OdeMapData message) { | ||
| try { | ||
| String time = ((OdeMapMetadata)message.getMetadata()).getOdeReceivedAt(); | ||
| return Instant.from(formatter.parse(time)); | ||
| } catch (Exception e) { | ||
| return Instant.ofEpochMilli(0); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isDuplicate(OdeMapData lastMessage, OdeMapData newMessage) { | ||
|
|
||
| Instant newValueTime = getMessageTime(newMessage); | ||
| Instant oldValueTime = getMessageTime(lastMessage); | ||
|
|
||
| if(newValueTime.minus(Duration.ofHours(1)).isAfter(oldValueTime)){ | ||
| return false; | ||
|
|
||
| }else{ | ||
| OdeMapPayload oldPayload = (OdeMapPayload)lastMessage.getPayload(); | ||
| OdeMapPayload newPayload = (OdeMapPayload)newMessage.getPayload(); | ||
|
|
||
| Integer oldTimestamp = oldPayload.getMap().getTimeStamp(); | ||
| Integer newTimestamp = newPayload.getMap().getTimeStamp(); | ||
|
|
||
|
|
||
| newPayload.getMap().setTimeStamp(oldTimestamp); | ||
|
|
||
| int oldHash = hashMapMessage(lastMessage); | ||
| int newhash = hashMapMessage(newMessage); | ||
|
|
||
| if(oldHash != newhash){ | ||
| newPayload.getMap().setTimeStamp(newTimestamp); | ||
| return false; | ||
| } | ||
| } | ||
| return true; | ||
| } | ||
|
|
||
| public int hashMapMessage(OdeMapData map){ | ||
| OdeMapPayload payload = (OdeMapPayload)map.getPayload(); | ||
| return Objects.hash(payload.toJson()); | ||
|
|
||
| } | ||
| } |
Oops, something went wrong.