Partition by intersection

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/AppHealthMonitor.java

@@ -43,11 +43,15 @@
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.AlgorithmParameters;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.StreamsTopology;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.config.ConfigParameters;
-import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmIntersectionKey;
+import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmIntersectionIdKey;
+import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmRsuIdKey;
 import us.dot.its.jpo.conflictmonitor.monitor.models.map.MapIndex;
 import us.dot.its.jpo.conflictmonitor.monitor.topologies.ConfigTopology;
 import us.dot.its.jpo.conflictmonitor.monitor.topologies.IntersectionEventTopology;
+import us.dot.its.jpo.conflictmonitor.monitor.utils.BsmUtils;
 import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey;
+import us.dot.its.jpo.geojsonconverter.pojos.geojson.LineString;
+import us.dot.its.jpo.geojsonconverter.pojos.geojson.map.ProcessedMap;
 import us.dot.its.jpo.geojsonconverter.pojos.spat.ProcessedSpat;
 import us.dot.its.jpo.ode.model.OdeBsmData;
 
@@ -104,7 +108,8 @@ public List<Object> parameterObjects() {
                     "connectors",
                     "spatial-indexes",
                     "spat-window-store",
-                    "bsm-window-store"
+                    "bsm-window-store",
+                    "map-store"
                 );
             return getJsonResponse(linkMap);
         } catch (Exception ex) {
@@ -256,6 +261,21 @@ private void addLinks(Map<String, String> map, String... paths) {
         return getJsonResponse(allItems);
     }
 
+    @GetMapping(value = "/map-store")
+    public @ResponseBody ResponseEntity<String> mapStore() {
+        var mapStore = intersectionEventTopology.getMapStore();
+        var mapMap = new TreeMap<String, ProcessedMap<LineString>>();
+        try (var mapIterator = mapStore.all()) {
+            while (mapIterator.hasNext()) {
+                var kvp = mapIterator.next();
+                RsuIntersectionKey key = kvp.key;
+                ProcessedMap<LineString> map = kvp.value;
+                mapMap.put(key.toString(), map);
+            }
+        }
+        return getJsonResponse(mapMap);
+    }
+
     @GetMapping(value = "/spat-window-store")
     public @ResponseBody ResponseEntity<String> spatWindowStore() {
         var spatWindowStore = intersectionEventTopology.getSpatWindowStore();
@@ -299,13 +319,13 @@ private void addLinks(Map<String, String> map, String... paths) {
         try (var bsmIterator = bsmWindowStore.all()) {
             while (bsmIterator.hasNext()) {
                 var kvp = bsmIterator.next();
-                Windowed<BsmIntersectionKey> key = kvp.key;
+                Windowed<BsmIntersectionIdKey> key = kvp.key;
                 Instant startTime = key.window().startTime();
                 Instant endTime = key.window().endTime();
-                BsmIntersectionKey theKey= key.key();
+                BsmIntersectionIdKey theKey= key.key();
                 OdeBsmData value = kvp.value;
                 // Integer intersectionId = value.();
-                String vehicleId = IntersectionEventTopology.getBsmID(value);
+                String vehicleId = BsmUtils.getVehicleId(value);
                 TreeMap<String, TreeMap<String, OdeBsmData>> bsms = null;
                 if (intersectionMap.containsKey(vehicleId)) {
                     bsms = intersectionMap.get(vehicleId);

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/bsm_event/BsmEventParameters.java

@@ -31,6 +31,11 @@ public class BsmEventParameters {
         updateType = READ_ONLY)
     String outputTopic;
 
+    @ConfigData(key = "bsm.event.bsmIntersectionOutputTopic",
+        description = "The Kafka topic to write BSMs partitioned by intersection to",
+        updateType = READ_ONLY)
+    String bsmIntersectionOutputTopic;
+
     @ConfigData(key = "bsm.event.stateStoreName", 
         description = "The name of the Timestamped KeyValue Store for BSMs",
         updateType = READ_ONLY)

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/intersection_event/IntersectionEventStreamsAlgorithm.java

@@ -1,11 +1,11 @@
 package us.dot.its.jpo.conflictmonitor.monitor.algorithms.intersection_event;
 
-import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
 import org.apache.kafka.streams.state.ReadOnlyWindowStore;
 
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.StreamsTopology;
-import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmIntersectionKey;
+import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmIntersectionIdKey;
+import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmRsuIdKey;
 import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey;
 import us.dot.its.jpo.geojsonconverter.pojos.geojson.LineString;
 import us.dot.its.jpo.geojsonconverter.pojos.geojson.map.ProcessedMap;
@@ -15,7 +15,7 @@
 public interface IntersectionEventStreamsAlgorithm 
     extends IntersectionEventAlgorithm, StreamsTopology { 
 
-    ReadOnlyWindowStore<BsmIntersectionKey, OdeBsmData> getBsmWindowStore();
+    ReadOnlyWindowStore<BsmIntersectionIdKey, OdeBsmData> getBsmWindowStore();
     ReadOnlyWindowStore<RsuIntersectionKey, ProcessedSpat> getSpatWindowStore();
     ReadOnlyKeyValueStore<RsuIntersectionKey, ProcessedMap<LineString>> getMapStore();
 

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/message_ingest/MessageIngestStreamsAlgorithm.java

@@ -5,7 +5,8 @@
 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
 import org.apache.kafka.streams.state.ReadOnlyWindowStore;
 
-import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmIntersectionKey;
+import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmIntersectionIdKey;
+import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmRsuIdKey;
 import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey;
 import us.dot.its.jpo.geojsonconverter.pojos.geojson.LineString;
 import us.dot.its.jpo.geojsonconverter.pojos.geojson.map.ProcessedMap;
@@ -15,7 +16,7 @@
 public interface MessageIngestStreamsAlgorithm
     extends MessageIngestAlgorithm {
 
-    ReadOnlyWindowStore<BsmIntersectionKey, OdeBsmData> getBsmWindowStore(KafkaStreams streams);
+    ReadOnlyWindowStore<BsmIntersectionIdKey, OdeBsmData> getBsmWindowStore(KafkaStreams streams);
     ReadOnlyWindowStore<RsuIntersectionKey, ProcessedSpat> getSpatWindowStore(KafkaStreams streams);
     ReadOnlyKeyValueStore<RsuIntersectionKey, ProcessedMap<LineString>> getMapStore(KafkaStreams streams);
 

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/bsm/BsmEvent.java

@@ -5,17 +5,15 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
-import lombok.EqualsAndHashCode;
-import lombok.Generated;
-import lombok.Getter;
-import lombok.Setter;
+import lombok.*;
 import us.dot.its.jpo.geojsonconverter.DateJsonMapper;
 import us.dot.its.jpo.geojsonconverter.pojos.geojson.LineString;
 import us.dot.its.jpo.ode.model.OdeBsmData;
 
 @Getter
 @Setter
 @EqualsAndHashCode
+@ToString
 @Generated
 public class BsmEvent {
     private OdeBsmData startingBsm;

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/bsm/BsmIntersectionIdKey.java

@@ -0,0 +1,36 @@
+package us.dot.its.jpo.conflictmonitor.monitor.models.bsm;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import us.dot.its.jpo.conflictmonitor.monitor.models.map.IntersectionRegion;
+import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey;
+
+@Getter
+@Setter
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+public class BsmIntersectionIdKey extends RsuIntersectionKey {
+
+    String bsmId;
+
+    public BsmIntersectionIdKey() {}
+
+    public BsmIntersectionIdKey(String bsmId, String rsuId, int intersectionId) {
+        super(rsuId, intersectionId);
+        this.bsmId = bsmId;
+    }
+
+    public BsmIntersectionIdKey(String bsmId, String rsuId, int intersectionId, int region) {
+        super(rsuId, intersectionId, region);
+        this.bsmId = bsmId;
+    }
+
+    @JsonIgnore
+    public IntersectionRegion getIntersectionRegion() {
+        return new IntersectionRegion(getIntersectionId(), getRegion());
+    }
+
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/bsm/BsmIntersectionKey.java → jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/bsm/BsmRsuIdKey.java

@@ -1,7 +1,5 @@
 package us.dot.its.jpo.conflictmonitor.monitor.models.bsm;
 
-import java.util.Objects;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -14,16 +12,16 @@
 @EqualsAndHashCode
 @Getter
 @Setter
-public class BsmIntersectionKey implements RsuIdKey {
+public class BsmRsuIdKey implements RsuIdKey {
 
-    private static final Logger logger = LoggerFactory.getLogger(BsmIntersectionKey.class);
+    private static final Logger logger = LoggerFactory.getLogger(BsmRsuIdKey.class);
 
     private String rsuId;
     private String bsmId;
 
-    public BsmIntersectionKey() {}
+    public BsmRsuIdKey() {}
 
-    public BsmIntersectionKey(String rsuId, String bsmId){
+    public BsmRsuIdKey(String rsuId, String bsmId){
         this.rsuId = rsuId;
         this.bsmId = bsmId;
     }

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/map/IntersectionRegion.java

@@ -11,8 +11,12 @@ public class IntersectionRegion {
     public IntersectionRegion() {}
 
     public IntersectionRegion(Integer intersectionId, Integer region) {
-        this.intersectionId = intersectionId;
-        this.region = region;
+        if (intersectionId != null && intersectionId.intValue() >= 0) {
+            this.intersectionId = intersectionId;
+        }
+        if (region != null && region.intValue() >= 0) {
+            this.region = region;
+        }
     }
 
     public IntersectionRegion(ProcessedMap map) {