Dynamic lane activation metrics

jpo-conflictmonitor/pom.xml

@@ -176,6 +176,12 @@
       <version>4.13.2</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>net.javacrumbs.json-unit</groupId>
+      <artifactId>json-unit</artifactId>
+      <version>5.1.0</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.jacoco</groupId>
       <artifactId>jacoco-maven-plugin</artifactId>

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

@@ -73,6 +73,8 @@
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.CommonMetricsParameters;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.priority_request.PriorityRequestMetricsAlgorithmFactory;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.priority_request.PriorityRequestMetricsParameters;
+import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation.DynamicLaneActivationMetricsAlgorithmFactory;
+import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation.DynamicLaneActivationMetricsParameters;
 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.message_ingest.MessageIngestAlgorithmFactory;
@@ -237,6 +239,9 @@ public class ConflictMonitorProperties implements EnvironmentAware  {
    private PriorityRequestMetricsAlgorithmFactory priorityRequestMetricsAlgorithmFactory;
    private String priorityRequestMetricsAlgorithm;
    private PriorityRequestMetricsParameters priorityRequestMetricsParameters;
+   private DynamicLaneActivationMetricsAlgorithmFactory dynamicLaneActivationMetricsAlgorithmFactory;
+   private String dynamicLaneActivationMetricsAlgorithm;
+   private DynamicLaneActivationMetricsParameters dynamicLaneActivationMetricsParameters;
 
    private NotificationAlgorithmFactory notificationAlgorithmFactory;
    private String notificationAlgorithm;
@@ -461,6 +466,12 @@ public void setPriorityRequestMetricsParameters(PriorityRequestMetricsParameters
       this.priorityRequestMetricsAlgorithm = priorityRequestMetricsParameters.getAlgorithm();
    }
 
+   @Autowired
+   public void setDynamicLaneActivationMetricsParameters(DynamicLaneActivationMetricsParameters dynamicLaneActivationMetricsParameters) {
+      this.dynamicLaneActivationMetricsParameters = dynamicLaneActivationMetricsParameters;
+      this.dynamicLaneActivationMetricsAlgorithm = dynamicLaneActivationMetricsParameters.getAlgorithm();
+   }
+
    @Autowired
    public void setCommonMetricsParameters(CommonMetricsParameters commonMetricsParameters) {
       this.commonMetricsParameters = commonMetricsParameters;
@@ -483,6 +494,12 @@ public void setPriorityRequestMetricsAlgorithmFactory(PriorityRequestMetricsAlgo
       this.priorityRequestMetricsAlgorithmFactory = factory;
    }
 
+   @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
+   @Autowired
+   public void setDynamicLaneActivationMetricsAlgorithmFactory(DynamicLaneActivationMetricsAlgorithmFactory factory) {
+      this.dynamicLaneActivationMetricsAlgorithmFactory = factory;
+   }
+
 
    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired

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

@@ -0,0 +1,4 @@
+package us.dot.its.jpo.conflictmonitor;
+
+public class Example {
+}

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

@@ -749,6 +749,15 @@ private BsmMessageCountProgressionAggregationAlgorithm getBsmMessageCountProgres
         return algorithm;
     }
 
+    private RevocableEnabledLaneAlignmentAggregationAlgorithm getRevocableEnabledLaneAlignmentAggregationAlgorithm() {
+        final var factory = conflictMonitorProps.getRevocableEnabledLaneAlignmentAggregationAlgorithmFactory();
+        final String algorithmName = conflictMonitorProps.getRevocableEnabledLaneAlignmentAggregationAlgorithm();
+        final var algorithm = factory.getAlgorithm(algorithmName);
+        final var parameters = conflictMonitorProps.getAggregationParameters();
+        algorithm.setParameters(parameters);
+        return algorithm;
+    }
+
     private RevocableEnabledLaneAlignmentAlgorithm getRevocableEnabledLaneAlignmentAlgorithm() {
         final var factory = conflictMonitorProps.getRevocableEnabledLaneAlignmentAlgorithmFactory();
         final String algorithmName = conflictMonitorProps.getRevocableEnabledLaneAlignmentAlgorithm();
@@ -758,15 +767,24 @@ private RevocableEnabledLaneAlignmentAlgorithm getRevocableEnabledLaneAlignmentA
         // Plug in aggregation algorithm
         final var aggAlgorithm = getRevocableEnabledLaneAlignmentAggregationAlgorithm();
         algorithm.setAggregationAlgorithm(aggAlgorithm);
+        // Plug in dynamic lane activation metrics algorithm
+        final var metricsAlgorithm = getDynamicLaneActivationMetricsAlgorithm();
+        algorithm.setDynamicLaneActivationMetricsAlgorithm(metricsAlgorithm);
         return algorithm;
     }
 
-    private RevocableEnabledLaneAlignmentAggregationAlgorithm getRevocableEnabledLaneAlignmentAggregationAlgorithm() {
-        final var factory = conflictMonitorProps.getRevocableEnabledLaneAlignmentAggregationAlgorithmFactory();
-        final String algorithmName = conflictMonitorProps.getRevocableEnabledLaneAlignmentAggregationAlgorithm();
+    private us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation.DynamicLaneActivationMetricsAlgorithm
+    getDynamicLaneActivationMetricsAlgorithm() {
+        final var factory = conflictMonitorProps.getDynamicLaneActivationMetricsAlgorithmFactory();
+        final String algorithmName = conflictMonitorProps.getDynamicLaneActivationMetricsAlgorithm();
         final var algorithm = factory.getAlgorithm(algorithmName);
-        final var parameters = conflictMonitorProps.getAggregationParameters();
+        final var parameters = conflictMonitorProps.getDynamicLaneActivationMetricsParameters();
         algorithm.setParameters(parameters);
+        final var commonParameters = conflictMonitorProps.getCommonMetricsParameters();
+        algorithm.setCommonParameters(commonParameters);
         return algorithm;
     }
-}
+}
+
+
+

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/metrics/MetricsStreamsAlgorithmInterface.java

@@ -6,10 +6,10 @@
 import us.dot.its.jpo.conflictmonitor.monitor.models.metrics.Metrics;
 
 /**
- * Streams implementation of a metrics algorithm that plugs into a topology, reads a stream of events,
+ * Streams implementation of a metrics algorithm that plugs into a topology, reads a stream of events or other messages,
  * and produces a stream of metrics.
  */
-public interface MetricsStreamsAlgorithmInterface<TEventKey, TMetricKey, TEvent extends Event, TMetric extends Metrics<TMetricKey>> {
+public interface MetricsStreamsAlgorithmInterface<TEventKey, TMetricKey, TEvent, TMetric extends Metrics<TMetricKey>> {
 
     KStream<TMetricKey, TMetric> buildTopology(StreamsBuilder builder, KStream<TEventKey, TEvent> inputStream);
 

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/metrics/dynamic_lane_activation/DynamicLaneActivationMetricsAlgorithm.java

@@ -0,0 +1,19 @@
+package us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation;
+
+import us.dot.its.jpo.conflictmonitor.monitor.algorithms.ConfigurableAlgorithm;
+import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.MetricsAlgorithmInterface;
+import us.dot.its.jpo.conflictmonitor.monitor.algorithms.revocable_enabled_lane_alignment.RevocableEnabledLaneAlignmentAlgorithm;
+import us.dot.its.jpo.conflictmonitor.monitor.models.metrics.dynamic_lane_activation.DynamicLaneActivationMetrics;
+import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey;
+
+/**
+ * Dynamic Lane Activation Metrics Algorithm
+ * <p>Tracks revocable lane enabled status</p>
+ * <p>Plugs into the {@link RevocableEnabledLaneAlignmentAlgorithm} as a subtopology</p>
+ */
+public interface DynamicLaneActivationMetricsAlgorithm
+    extends MetricsAlgorithmInterface<
+        DynamicLaneActivationMetrics,
+        RsuIntersectionKey,
+        DynamicLaneActivationMetricsParameters> {
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/metrics/dynamic_lane_activation/DynamicLaneActivationMetricsAlgorithmFactory.java

@@ -0,0 +1,5 @@
+package us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation;
+
+public interface DynamicLaneActivationMetricsAlgorithmFactory {
+    DynamicLaneActivationMetricsAlgorithm getAlgorithm(String algorithmName);
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/metrics/dynamic_lane_activation/DynamicLaneActivationMetricsAlgorithms.java

@@ -0,0 +1,16 @@
+package us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.config.ServiceLocatorFactoryBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class DynamicLaneActivationMetricsAlgorithms {
+    @Bean
+    public FactoryBean<?> dynamicLaneActivationMetricsServiceLocatorFactoryBean() {
+        var factoryBean = new ServiceLocatorFactoryBean();
+        factoryBean.setServiceLocatorInterface(DynamicLaneActivationMetricsAlgorithmFactory.class);
+        return factoryBean;
+    }
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/metrics/dynamic_lane_activation/DynamicLaneActivationMetricsConstants.java

@@ -0,0 +1,5 @@
+package us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation;
+
+public class DynamicLaneActivationMetricsConstants {
+    public static final String DEFAULT_DYNAMIC_LANE_ACTIVATION_METRICS_ALGORITHM = "defaultDynamicLaneActivationMetricsAlgorithm";
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/metrics/dynamic_lane_activation/DynamicLaneActivationMetricsParameters.java

@@ -0,0 +1,18 @@
+package us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation;
+
+import lombok.Data;
+import lombok.Generated;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+import us.dot.its.jpo.conflictmonitor.monitor.models.config.ConfigDataClass;
+
+@Data
+@Generated
+@Component
+@ConfigurationProperties(prefix = "metrics.dynamic.lane.activation")
+@ConfigDataClass
+public class DynamicLaneActivationMetricsParameters {
+    private String algorithm;
+    private volatile boolean debug;
+    private String outputMetricTopic;
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/metrics/dynamic_lane_activation/DynamicLaneActivationMetricsStreamsAlgorithm.java

@@ -0,0 +1,15 @@
+package us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation;
+
+import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.MetricsStreamsAlgorithmInterface;
+import us.dot.its.jpo.conflictmonitor.monitor.models.events.revocable_enabled_lane_alignment.RevocableLaneStatus;
+import us.dot.its.jpo.conflictmonitor.monitor.models.metrics.dynamic_lane_activation.DynamicLaneActivationMetrics;
+import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey;
+
+public interface DynamicLaneActivationMetricsStreamsAlgorithm
+    extends DynamicLaneActivationMetricsAlgorithm,
+        MetricsStreamsAlgorithmInterface<
+                RsuIntersectionKey,
+                RsuIntersectionKey,
+                RevocableLaneStatus,
+                DynamicLaneActivationMetrics> {
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/algorithms/revocable_enabled_lane_alignment/RevocableEnabledLaneAlignmentAlgorithm.java

@@ -2,10 +2,12 @@
 
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.ConfigurableAlgorithm;
 import us.dot.its.jpo.conflictmonitor.monitor.algorithms.aggregation.revocable_enabled_lane_alignment.RevocableEnabledLaneAlignmentAggregationAlgorithm;
+import us.dot.its.jpo.conflictmonitor.monitor.algorithms.metrics.dynamic_lane_activation.DynamicLaneActivationMetricsAlgorithm;
 
 public interface RevocableEnabledLaneAlignmentAlgorithm
     extends ConfigurableAlgorithm<RevocableEnabledLaneAlignmentParameters> {
 
     void setAggregationAlgorithm(RevocableEnabledLaneAlignmentAggregationAlgorithm aggregationAlgorithm);
 
+    void setDynamicLaneActivationMetricsAlgorithm(DynamicLaneActivationMetricsAlgorithm dynamicLaneActivationMetricsAlgorithm);
 }

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/events/revocable_enabled_lane_alignment/RevocableEnabledLaneAlignmentEvent.java

@@ -35,7 +35,7 @@ public RevocableEnabledLaneAlignmentEvent() {
     private LaneTypeAttributesMap laneTypeAttributes;
 
     /**
-     * Set of LanIDs with the 'revocable' bit set in the MAP message.
+     * Set of LaneIDs with the 'revocable' bit set in the MAP message.
      */
     private Set<Integer> revocableLaneList;
 

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/events/revocable_enabled_lane_alignment/RevocableLaneStatus.java

@@ -0,0 +1,67 @@
+package us.dot.its.jpo.conflictmonitor.monitor.models.events.revocable_enabled_lane_alignment;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.Generated;
+import lombok.NoArgsConstructor;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Lists all the revocable lanes for an intersection and their enabled/disabled status
+ * for a particular SPAT timestamp.
+ * This type is produced for all revocable enabled lane status changes and used for the
+ * Dynamic Lane Activation Metric.
+ */
+@Data
+@NoArgsConstructor
+@Generated
+public class RevocableLaneStatus {
+
+    public RevocableLaneStatus(RevocableEnabledLaneAlignmentEvent event) {
+        this.source = event.getSource();
+        this.intersectionID = event.getIntersectionID();
+        this.roadRegulatorID = event.getRoadRegulatorID();
+        this.timestamp = event.getTimestamp();
+        this.revocableLaneList = event.getRevocableLaneList();
+        this.enabledLaneList = event.getEnabledLaneList();
+    }
+
+    /**
+     * The SPAT source (usually IP address)
+     */
+    private String source;
+
+    /**
+     * The Intersection ID
+     */
+    private int intersectionID = -1;
+
+    /**
+     * The Road Regulator ID
+     */
+    private int roadRegulatorID = -1;
+
+    /**
+     * The timestamp of the SPAT
+     */
+    private long timestamp;
+
+    /**
+     * Set of LaneIDs with the 'revocable' bit set in the MAP message.
+     */
+    private Set<Integer> revocableLaneList = new HashSet<>();
+
+    /**
+     * Set of enabled Lane IDs from the SPAT message.
+     */
+    private Set<Integer> enabledLaneList = new HashSet<>();
+
+    /**
+     * Flag to indicate it this is a "tick" in the processing topology, not an actual status.
+     */
+    private boolean tick = false;
+
+
+}

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/metrics/Metrics.java

@@ -8,6 +8,7 @@
 import lombok.Generated;
 import lombok.extern.slf4j.Slf4j;
 import us.dot.its.jpo.conflictmonitor.monitor.models.events.ProcessingTimePeriod;
+import us.dot.its.jpo.conflictmonitor.monitor.models.metrics.dynamic_lane_activation.DynamicLaneActivationMetrics;
 import us.dot.its.jpo.geojsonconverter.DateJsonMapper;
 
 /**
@@ -23,7 +24,7 @@
         @Type(value = PriorityRequestMetrics.class,
                 name = "PriorityRequest"),
         @Type(value = DynamicLaneActivationMetrics.class,
-            name = "DynamicLaneActivationMetric")
+            name = "DynamicLaneActivation")
 })
 @Data
 @Generated

jpo-conflictmonitor/src/main/java/us/dot/its/jpo/conflictmonitor/monitor/models/metrics/dynamic_lane_activation/DynamicLaneActivationMetrics.java

@@ -0,0 +1,68 @@
+package us.dot.its.jpo.conflictmonitor.monitor.models.metrics.dynamic_lane_activation;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import us.dot.its.jpo.conflictmonitor.monitor.models.events.revocable_enabled_lane_alignment.LaneTypeAttributesMap;
+import us.dot.its.jpo.conflictmonitor.monitor.models.metrics.Metrics;
+import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey;
+
+/**
+ * Performance and Operational Metrics: Dynamic Lane Activation Monitoring
+ * Object Tracks which revocable lanes are active per rsu/intersection (for intersections
+ * that have them).
+ * <p>Notes:</p>
+ * <ul>
+ *     <li>The time period for the metric corresponds to timestamps of SPATs that were received for which the set of
+ *     enabled revocable lanes changed.</li>
+ *      <li>This metric is only produced when the enabled status of any revocable lane in the intersection changes.</li>
+ *      <li>This metric is <b>not</b> produced for every SPAT message.</li>
+ *      <li>This metric is <b>not</b> produced for intersections that don't have any revocable lanes (and most
+ *      intersections generally don't have any)</li>
+ * </ul>
+ */
+@Getter
+@Setter
+@EqualsAndHashCode(callSuper = true)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DynamicLaneActivationMetrics extends Metrics<RsuIntersectionKey> {
+    public DynamicLaneActivationMetrics() {
+        super("DynamicLaneActivation");
+    }
+
+    /**
+     * The SPAT source (usually the IP address of the RSU)
+     */
+    @JsonProperty(access = JsonProperty.Access.READ_ONLY)
+    private String getSource() {
+        return key != null ? key.getRsuId() : null;
+    }
+
+    /**
+     * The intersection ID
+     */
+    @JsonProperty(access = JsonProperty.Access.READ_ONLY)
+    private int getIntersectionID() {
+        return key != null ? key.getIntersectionId() : -1;
+    }
+
+    /**
+     * The RoadRegulatorID (region)
+     */
+    @JsonProperty(access = JsonProperty.Access.READ_ONLY)
+    private int getRoadRegulatorID() {
+        return key != null ? key.getRegion() : -1;
+    }
+
+    /**
+     * Table of the enabled status of each revocable lane in the intersection of the time period
+     * of the metric.
+     */
+    private RevocableEnabledLaneStatusTable revocableEnabledLaneStatusTable = new RevocableEnabledLaneStatusTable();
+
+}
+
+