Integrate ProcessedBsm

docker-compose-ode.yml

@@ -7,7 +7,7 @@ services:
       - all
       - cm_full
       - ode_geojsonconverter
-    image: usdotjpoode/geojsonconverter:latest
+    image: usdotjpoode/geojsonconverter:2.0.0
     restart: ${RESTART_POLICY}
     deploy:
       resources:
@@ -36,7 +36,7 @@ services:
       - all
       - cm_full
       - ode_geojsonconverter
-    image: usdotjpoode/jpo-ode:latest
+    image: usdotjpoode/jpo-ode:4.0.0
     restart: ${RESTART_POLICY}
     deploy:
       resources:
@@ -90,7 +90,7 @@ services:
       - all
       - cm_full
       - ode_geojsonconverter
-    image: usdotjpoode/asn1_codec:latest
+    image: usdotjpoode/asn1_codec:3.0.0
     restart: ${RESTART_POLICY}
     deploy:
       resources:

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/ConflictMonitorProperties.java

@@ -75,8 +75,6 @@
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.notification.NotificationParameters;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.message_ingest.MessageIngestAlgorithmFactory;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.message_ingest.MessageIngestParameters;
-import us.dot.its.jpo.conflictmonitor.monitor.algorithms.repartition.RepartitionAlgorithmFactory;
-import us.dot.its.jpo.conflictmonitor.monitor.algorithms.repartition.RepartitionParameters;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.map_message_count_progression.MapMessageCountProgressionAlgorithmFactory;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.map_message_count_progression.MapMessageCountProgressionParameters;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.spat_message_count_progression.SpatMessageCountProgressionAlgorithmFactory;
@@ -208,10 +206,6 @@ public class ConflictMonitorProperties implements EnvironmentAware  {
    private String connectionOfTravelAssessmentAlgorithm;
    private ConnectionOfTravelAssessmentParameters connectionOfTravelAssessmentAlgorithmParameters;
 
-   private RepartitionAlgorithmFactory repartitionAlgorithmFactory;
-   private String repartitionAlgorithm;
-   private RepartitionParameters repartitionAlgorithmParameters;
-
    private MapMessageCountProgressionAlgorithmFactory mapMessageCountProgressionAlgorithmFactory;
    private String mapMessageCountProgressionAlgorithm;
    private MapMessageCountProgressionParameters mapMessageCountProgressionAlgorithmParameters;
@@ -257,8 +251,7 @@ public class ConflictMonitorProperties implements EnvironmentAware  {
    private MessageIngestAlgorithmFactory messageIngestAlgorithmFactory;
    private MessageIngestParameters messageIngestParameters;
 
-
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setAggregationParameters(AggregationParameters aggregationParameters) {
       this.aggregationParameters = aggregationParameters;
@@ -375,11 +368,13 @@ public void setSpatValidationParameters(SpatValidationParameters spatBroadcastRa
       this.spatValidationParameters = spatBroadcastRateParameters;
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setMapValidationAlgorithmFactory(MapValidationAlgorithmFactory factory) {
       this.mapValidationAlgorithmFactory = factory;
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSpatValidationAlgorithmFactory(SpatValidationStreamsAlgorithmFactory factory) {
       this.spatValidationAlgorithmFactory = factory;
@@ -402,7 +397,7 @@ public void setMapTimestampDeltaParameters(MapTimestampDeltaParameters mapTimest
       this.mapTimestampDeltaAlgorithm = mapTimestampDeltaParameters.getAlgorithm();
    }
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setMapTimestampDeltaAlgorithmFactory(MapTimestampDeltaAlgorithmFactory factory) {
       this.mapTimestampDeltaAlgorithmFactory = factory;
@@ -415,14 +410,15 @@ public void setSpatTimestampDeltaParameters(SpatTimestampDeltaParameters spatTim
       this.spatTimestampDeltaAlgorithm = spatTimestampDeltaParameters.getAlgorithm();
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSpatTimestampDeltaAlgorithmFactory(SpatTimestampDeltaAlgorithmFactory factory) {
       this.spatTimestampDeltaAlgorithmFactory = factory;
    }
 
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setLaneDirectionOfTravelAlgorithmFactory(
          LaneDirectionOfTravelAlgorithmFactory laneDirectionOfTravelAlgorithmFactory) {
@@ -444,7 +440,7 @@ public void setLaneDirectionOfTravelParameters(LaneDirectionOfTravelParameters l
    }
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setConnectionOfTravelAlgorithmFactory(
          ConnectionOfTravelAlgorithmFactory connectionOfTravelAlgorithmFactory) {
@@ -466,7 +462,7 @@ public void setConnectionOfTravelParameters(ConnectionOfTravelParameters connect
    }
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSignalStateVehicleCrossesAlgorithmFactory(
          StopLinePassageAlgorithmFactory signalStateVehicleCrossesAlgorithmFactory) {
@@ -489,7 +485,7 @@ public void setSignalStateVehicleCrossesParameters(
    }
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSignalStateVehicleStopsAlgorithmFactory(
          StopLineStopAlgorithmFactory signalStateVehicleStopsAlgorithmFactory) {
@@ -511,7 +507,7 @@ public void setSignalStateVehicleStopsParameters(StopLineStopParameters signalSt
    }
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setMapSpatMessageAssessmentAlgorithmFactory(
          MapSpatMessageAssessmentAlgorithmFactory mapSpatMessageAssessmentAlgorithmFactory) {
@@ -533,7 +529,7 @@ public void setMapSpatMessageAssessmentParameters(
 
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSpatTimeChangeDetailsAlgorithmFactory(
          SpatTimeChangeDetailsAlgorithmFactory spatTimeChangeDetailsAlgorithmFactory) {
@@ -561,6 +557,7 @@ public void setSpatTimeChangeDetailsParameters(SpatTimeChangeDetailsParameters s
       this.spatTimeChangeDetailsParameters = spatTimeChangeDetailsParameters;
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSpatTransitionAlgorithmFactory(EventStateProgressionAlgorithmFactory spatTransitionAlgorithmFactory) {
       this.spatTransitionAlgorithmFactory = spatTransitionAlgorithmFactory;
@@ -576,7 +573,7 @@ public void setSpatTransitionParameters(EventStateProgressionParameters spatTran
       this.spatTransitionParameters = spatTransitionParameters;
    }
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setMapTimeChangeDetailsAlgorithmFactory(
          MapTimeChangeDetailsAlgorithmFactory mapTimeChangeDetailsAlgorithmFactory) {
@@ -599,7 +596,7 @@ public void setMapTimeChangeDetailsParameters(MapTimeChangeDetailsParameters map
 
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSignalStateEventAssessmentAlgorithmFactory(
          StopLinePassageAssessmentAlgorithmFactory signalStateEventAssessmentAlgorithmFactory) {
@@ -621,6 +618,7 @@ public void setSignalStateEventAssessmentAlgorithmParameters(
       this.signalStateEventAssessmentAlgorithmParameters = signalStateEventAssessmentAlgorithmParameters;
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setStopLineStopAssessmentAlgorithmFactory(
          StopLineStopAssessmentAlgorithmFactory stopLineStopAssessmentAlgorithmFactory) {
@@ -643,7 +641,7 @@ public void setStopLineStopAssessmentAlgorithmParameters(
    }
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setLaneDirectionOfTravelAssessmentAlgorithmFactory(
          LaneDirectionOfTravelAssessmentAlgorithmFactory laneDirectionfOfTravelAssessmentAlgorithmFactory) {
@@ -666,7 +664,7 @@ public void setLaneDirectionOfTravelAssessmentAlgorithmParameters(
    }
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setConnectionOfTravelAssessmentAlgorithmFactory(
          ConnectionOfTravelAssessmentAlgorithmFactory connectionOfTravelAssessmentAlgorithmFactory) {
@@ -688,28 +686,7 @@ public void setConnectionOfTravelAssessmentAlgorithmParameters(
       this.connectionOfTravelAssessmentAlgorithmParameters = connectionOfTravelAssessmentAlgorithmParameters;
    }
 
-
-
-   @Autowired
-   public void setRepartitionAlgorithmFactory(RepartitionAlgorithmFactory repartitionAlgorithmFactory) {
-      this.repartitionAlgorithmFactory = repartitionAlgorithmFactory;
-   }
-
-
-
-   @Value("${repartition.algorithm}")
-   public void setRepartitionAlgorithm(String repartitionAlgorithm) {
-      this.repartitionAlgorithm = repartitionAlgorithm;
-   }
-
-
-
-   @Autowired
-   public void setRepartitionAlgorithmParameters(RepartitionParameters repartitionAlgorithmParameters) {
-      this.repartitionAlgorithmParameters = repartitionAlgorithmParameters;
-   }
-
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setMapMessageCountProgressionAlgorithmFactory(MapMessageCountProgressionAlgorithmFactory mapMessageCountProgressionAlgorithmFactory) {
       this.mapMessageCountProgressionAlgorithmFactory = mapMessageCountProgressionAlgorithmFactory;
@@ -726,7 +703,7 @@ public void setMapMessageCountProgressionAlgorithmParameters(MapMessageCountProg
    }
 
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setSpatMessageCountProgressionAlgorithmFactory(SpatMessageCountProgressionAlgorithmFactory spatMessageCountProgressionAlgorithmFactory) {
       this.spatMessageCountProgressionAlgorithmFactory = spatMessageCountProgressionAlgorithmFactory;
@@ -742,7 +719,7 @@ public void setSpatMessageCountProgressionAlgorithmParameters(SpatMessageCountPr
       this.spatMessageCountProgressionAlgorithmParameters = spatMessageCountProgressionAlgorithmParameters;
    }
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setBsmMessageCountProgressionAlgorithmFactory(BsmMessageCountProgressionAlgorithmFactory bsmMessageCountProgressionAlgorithmFactory) {
       this.bsmMessageCountProgressionAlgorithmFactory = bsmMessageCountProgressionAlgorithmFactory;
@@ -763,9 +740,9 @@ public NotificationAlgorithmFactory getNotificationAlgorithmFactory() {
       return notificationAlgorithmFactory;
    }
 
-
-
 
+
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setNotificationAlgorithmFactory(NotificationAlgorithmFactory notificationAlgorithmFactory) {
       this.notificationAlgorithmFactory = notificationAlgorithmFactory;
@@ -795,6 +772,7 @@ public void setEventParameters(EventParameters eventParameters) {
       this.eventAlgorithm = eventParameters.getAlgorithm();
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setEventAlgorithmFactory(EventAlgorithmFactory factory) {
       this.eventAlgorithmFactory = factory;
@@ -804,7 +782,7 @@ public Boolean getConfluentCloudStatus() {
 		return confluentCloudEnabled;
 	}
 
-
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setBsmEventAlgorithmFactory(BsmEventAlgorithmFactory bsmEventAlgorithmFactory) {
       this.bsmEventAlgorithmFactory = bsmEventAlgorithmFactory;
@@ -815,6 +793,7 @@ public void setBsmEventParameters(BsmEventParameters bsmEventParameters) {
       this.bsmEventParameters = bsmEventParameters;
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    public void setMessageIngestAlgorithmFactory(MessageIngestAlgorithmFactory messageIngestAlgorithmFactory) {
       this.messageIngestAlgorithmFactory = messageIngestAlgorithmFactory;

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

@@ -27,7 +27,6 @@
 import org.springframework.web.client.RestTemplate;
 import org.springframework.web.servlet.support.ServletUriComponentsBuilder;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
@@ -49,9 +48,10 @@
 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.Point;
+import us.dot.its.jpo.geojsonconverter.pojos.geojson.bsm.ProcessedBsm;
 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;
 
 import javax.ws.rs.Produces;
 
@@ -360,22 +360,22 @@ private void addLinks(Map<String, String> map, String... paths) {
                 Instant startTime = key.window().startTime();
                 Instant endTime = key.window().endTime();
                 BsmIntersectionIdKey theKey= key.key();
-                OdeBsmData value = kvp.value;
+                ProcessedBsm<Point> value = kvp.value;
                 // Integer intersectionId = value.();
                 String vehicleId = BsmUtils.getVehicleId(value);
-                TreeMap<String, TreeMap<String, OdeBsmData>> bsms = null;
+                TreeMap<String, TreeMap<String, ProcessedBsm<Point>>> bsms = null;
                 if (intersectionMap.containsKey(vehicleId)) {
                     bsms = intersectionMap.get(vehicleId);
                 } else {
-                    bsms = new TreeMap<String, TreeMap<String, OdeBsmData>>();
+                    bsms = new TreeMap<>();
                     intersectionMap.put(vehicleId, bsms);
                 }
                 String window = String.format("%s / %s", formatter.format(startTime.atZone(ZoneOffset.UTC)), formatter.format(endTime.atZone(ZoneOffset.UTC)));
-                TreeMap<String, OdeBsmData> bsmList = null;
+                TreeMap<String, ProcessedBsm<Point>> bsmList = null;
                 if (bsms.containsKey(window)) {
                     bsmList = bsms.get(window);
                 } else {
-                    bsmList = new TreeMap<String, OdeBsmData>();
+                    bsmList = new TreeMap<>();
                     bsms.put(window, bsmList);
                 }
                 bsmList.put(theKey.toString(), value);
@@ -458,7 +458,7 @@ private String baseUrl() {
     }
 
     public class IntersectionSpatMap extends TreeMap<Integer, TreeMap<String, TreeMap<String, ProcessedSpat>>> {}
-    public class IntersectionBsm extends TreeMap<String, TreeMap<String, TreeMap<String, OdeBsmData>>> {}
+    public class IntersectionBsm extends TreeMap<String, TreeMap<String, TreeMap<String, ProcessedBsm<Point>>>> {}
 
 
     public record TopologyInfoLinks(String detailsUrl, String simpleGraphUrl) {}

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

@@ -54,9 +54,6 @@
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.notification.NotificationAlgorithm;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.notification.NotificationAlgorithmFactory;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.notification.NotificationParameters;
-import us.dot.its.jpo.conflictmonitor.monitor.algorithms.repartition.RepartitionAlgorithm;
-import us.dot.its.jpo.conflictmonitor.monitor.algorithms.repartition.RepartitionAlgorithmFactory;
-import us.dot.its.jpo.conflictmonitor.monitor.algorithms.repartition.RepartitionParameters;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.event_state_progression.EventStateProgressionAlgorithm;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.stop_line_passage.StopLinePassageAlgorithm;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.stop_line_passage.StopLinePassageAlgorithmFactory;
@@ -153,24 +150,6 @@ public MonitorServiceController(final ConflictMonitorProperties conflictMonitorP
             configTopology.start();
 
 
-            final String repartition = "repartition";
-            final RepartitionAlgorithmFactory repartitionAlgoFactory = conflictMonitorProps.getRepartitionAlgorithmFactory();
-            final String repAlgo = conflictMonitorProps.getRepartitionAlgorithm();
-            final RepartitionAlgorithm repartitionAlgo = repartitionAlgoFactory.getAlgorithm(repAlgo);
-            final RepartitionParameters repartitionParams = conflictMonitorProps.getRepartitionAlgorithmParameters();
-            configTopology.registerConfigListeners(repartitionParams);
-            if (repartitionAlgo instanceof StreamsTopology) {     
-                final var streamsAlgo = (StreamsTopology)repartitionAlgo;
-                streamsAlgo.setStreamsProperties(conflictMonitorProps.createStreamProperties(repartition));
-                streamsAlgo.registerStateListener(new StateChangeHandler(kafkaTemplate, repartition, stateChangeTopic, healthTopic));
-                streamsAlgo.registerUncaughtExceptionHandler(new StreamsExceptionHandler(kafkaTemplate, repartition, healthTopic));
-                algoMap.put(repartition, streamsAlgo);
-            }
-            repartitionAlgo.setParameters(repartitionParams);
-            Runtime.getRuntime().addShutdownHook(new Thread(repartitionAlgo::stop));
-            repartitionAlgo.start();
-
-
             final String notification = "notification";
             final NotificationAlgorithmFactory notificationAlgoFactory = conflictMonitorProps.getNotificationAlgorithmFactory();
             final String notAlgo = conflictMonitorProps.getNotificationAlgorithm();

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/aggregation/bsm_message_count_progression/BsmMessageCountProgressionAggregationKey.java

@@ -1,13 +1,13 @@
 package us.dot.its.jpo.conflictmonitor.monitor.algorithms.aggregation.bsm_message_count_progression;
 
 import lombok.*;
-import us.dot.its.jpo.conflictmonitor.monitor.models.bsm.BsmRsuIdKey;
-import us.dot.its.jpo.geojsonconverter.partitioner.RsuIdKey;
+
+import us.dot.its.jpo.geojsonconverter.partitioner.RsuLogKey;
 
 @EqualsAndHashCode(callSuper = true)
 @Data
 @Generated
-public class BsmMessageCountProgressionAggregationKey extends BsmRsuIdKey {
+public class BsmMessageCountProgressionAggregationKey extends RsuLogKey {
     private String dataFrame;
     private String change;
 }