feat: add MeasureProcessor to process full measure (such as total)

Fixes #160

Author: metacosm

analysis/src/test/java/ComputeTest.java

@@ -78,9 +78,12 @@ void averageShouldWork() {
         final var m3c3 = 0.0;
 
         final var measure = new OngoingPowerMeasure(metadata);
-        measure.registerProcessorFor(0, new MeanComponentProcessor());
-        measure.registerProcessorFor(1, new MeanComponentProcessor());
-        measure.registerProcessorFor(2, new MeanComponentProcessor());
+        final var avgProc1 = new MeanComponentProcessor();
+        measure.registerProcessorFor(0, avgProc1);
+        final var avgProc2 = new MeanComponentProcessor();
+        measure.registerProcessorFor(1, avgProc2);
+        final var avgProc3 = new MeanComponentProcessor();
+        measure.registerProcessorFor(2, avgProc3);
 
         final var components = new double[metadata.componentCardinality()];
         components[0] = m1c1;
@@ -104,17 +107,13 @@ void averageShouldWork() {
 
         assertEquals((m1c1 + m2c1 + m3c1) / 3, c1Avg, 0.0001,
                 "Average did not match the expected value");
-        final var processors = measure.processors();
-        assertEquals(c1Avg,
-                processors.processorFor(0, MeanComponentProcessor.class).map(MeanComponentProcessor::mean).orElseThrow());
+        assertEquals(c1Avg, avgProc1.mean());
         assertEquals((m1c2 + m2c2 + m3c2) / 3, c2Avg, 0.0001,
                 "Average did not match the expected value");
-        assertEquals(c2Avg,
-                processors.processorFor(1, MeanComponentProcessor.class).map(MeanComponentProcessor::mean).orElseThrow());
+        assertEquals(c2Avg, avgProc2.mean());
         assertEquals(0, c3Avg, 0.0001,
                 "Average did not match the expected value");
-        assertEquals(0,
-                processors.processorFor(2, MeanComponentProcessor.class).map(MeanComponentProcessor::mean).orElseThrow());
+        assertEquals(0, avgProc3.mean());
 
         System.out.println(PowerMeasure.asString(measure));
     }

measure/src/main/java/net/laprun/sustainability/power/analysis/ComponentProcessor.java

@@ -1,14 +1,6 @@
 package net.laprun.sustainability.power.analysis;
 
-public interface ComponentProcessor {
+public interface ComponentProcessor extends Outputable {
     default void recordComponentValue(double value, long timestamp) {
     }
-
-    default String name() {
-        return this.getClass().getSimpleName();
-    }
-
-    default String output() {
-        return "";
-    }
 }

measure/src/main/java/net/laprun/sustainability/power/analysis/DefaultProcessors.java

@@ -3,13 +3,12 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
-import java.util.Optional;
 
 import net.laprun.sustainability.power.SensorMetadata;
 
 public class DefaultProcessors implements Processors {
     private final List<ComponentProcessor>[] processors;
-    private ComponentProcessor totalProcessor;
+    private List<MeasureProcessor> measureProcessors;
 
     @SuppressWarnings("unchecked")
     public DefaultProcessors(int componentCardinality) {
@@ -18,6 +17,10 @@ public DefaultProcessors(int componentCardinality) {
 
     @Override
     public void recordMeasure(double[] components, long timestamp) {
+        if (measureProcessors != null) {
+            measureProcessors.forEach(processor -> processor.recordMeasure(components, timestamp));
+        }
+
         for (var index = 0; index < components.length; index++) {
             final var fIndex = index;
             final var componentProcessors = processors[index];
@@ -39,43 +42,31 @@ public void registerProcessorFor(int componentIndex, ComponentProcessor processo
         }
     }
 
-    @Override
-    public void registerTotalProcessor(ComponentProcessor processor) {
-        this.totalProcessor = processor;
-    }
-
-    @Override
-    public void recordTotal(double total, long timestamp) {
-        if (totalProcessor != null) {
-            totalProcessor.recordComponentValue(total, timestamp);
-        }
-    }
-
-    @Override
-    public Optional<ComponentProcessor> totalProcessor() {
-        return Optional.ofNullable(totalProcessor);
-    }
-
     @Override
     public List<ComponentProcessor> processorsFor(int componentIndex) {
         final var componentProcessors = processors[componentIndex];
         return Objects.requireNonNullElseGet(componentProcessors, List::of);
     }
 
     @Override
-    public <T extends ComponentProcessor> Optional<T> processorFor(int componentIndex, Class<T> expectedType) {
-        final var componentProcessors = processors[componentIndex];
-        if (componentProcessors != null) {
-            return componentProcessors.stream().filter(expectedType::isInstance).map(expectedType::cast).findFirst();
-        } else {
-            return Optional.empty();
+    public void registerMeasureProcessor(MeasureProcessor processor) {
+        if (processor != null) {
+            if (measureProcessors == null) {
+                measureProcessors = new ArrayList<>();
+            }
+            measureProcessors.add(processor);
         }
     }
 
+    @Override
+    public List<MeasureProcessor> measureProcessors() {
+        return Objects.requireNonNullElseGet(measureProcessors, List::of);
+    }
+
     @Override
     public String output(SensorMetadata metadata) {
         StringBuilder builder = new StringBuilder();
-        builder.append("# Processors\n");
+        builder.append("# Component Processors\n");
         for (int i = 0; i < processors.length; i++) {
             final var name = metadata.metadataFor(i).name();
             builder.append("  - ").append(name).append(" component:\n");

measure/src/main/java/net/laprun/sustainability/power/analysis/MeasureProcessor.java

@@ -0,0 +1,6 @@
+package net.laprun.sustainability.power.analysis;
+
+public interface MeasureProcessor extends Outputable {
+    default void recordMeasure(double[] measure, long timestamp) {
+    }
+}

measure/src/main/java/net/laprun/sustainability/power/analysis/Outputable.java

@@ -0,0 +1,11 @@
+package net.laprun.sustainability.power.analysis;
+
+public interface Outputable {
+    default String name() {
+        return this.getClass().getSimpleName();
+    }
+
+    default String output() {
+        return "";
+    }
+}

measure/src/main/java/net/laprun/sustainability/power/analysis/Processors.java

@@ -1,38 +1,31 @@
 package net.laprun.sustainability.power.analysis;
 
 import java.util.List;
-import java.util.Optional;
 
 import net.laprun.sustainability.power.SensorMetadata;
 
+@SuppressWarnings("unused")
 public interface Processors {
     Processors empty = new Processors() {
     };
 
     default void recordMeasure(double[] components, long timestamp) {
     }
 
-    default void recordTotal(double total, long timestamp) {
-    }
-
     default void registerProcessorFor(int componentIndex, ComponentProcessor processor) {
     }
 
-    default void registerTotalProcessor(ComponentProcessor processor) {
+    default List<ComponentProcessor> processorsFor(int componentIndex) {
+        return List.of();
     }
 
-    default Optional<ComponentProcessor> totalProcessor() {
-        return Optional.empty();
+    default void registerMeasureProcessor(MeasureProcessor processor) {
     }
 
-    default List<ComponentProcessor> processorsFor(int componentIndex) {
+    default List<MeasureProcessor> measureProcessors() {
         return List.of();
     }
 
-    default <T extends ComponentProcessor> Optional<T> processorFor(int componentIndex, Class<T> expectedType) {
-        return Optional.empty();
-    }
-
     default String output(SensorMetadata metadata) {
         return "";
     }

measure/src/main/java/net/laprun/sustainability/power/analysis/TotalMeasureProcessor.java

@@ -0,0 +1,90 @@
+package net.laprun.sustainability.power.analysis;
+
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import net.laprun.sustainability.power.Errors;
+import net.laprun.sustainability.power.SensorMetadata;
+import net.laprun.sustainability.power.SensorUnit;
+
+public class TotalMeasureProcessor implements MeasureProcessor {
+    private final String name;
+    private double minTotal = Double.MAX_VALUE;
+    private double maxTotal;
+    private double accumulatedTotal;
+    private final Function<double[], Double> formula;
+
+    public TotalMeasureProcessor(SensorMetadata metadata, int... totalComponentIndices) {
+        Objects.requireNonNull(totalComponentIndices, "Must specify component indices that will aggregated in a total");
+
+        final var errors = new Errors();
+        final var totalComponents = Arrays.stream(totalComponentIndices)
+                .mapToObj(i -> toTotalComponent(metadata, i, errors))
+                .toArray(TotalComponent[]::new);
+        name = Arrays.stream(totalComponents)
+                .map(TotalComponent::name)
+                .collect(Collectors.joining(" + ", "Aggregated total from (", ")"));
+        formula = components -> {
+            double result = 0;
+            for (var totalComponent : totalComponents) {
+                result += components[totalComponent.index] * totalComponent.factor;
+            }
+            return result;
+        };
+
+        if (errors.hasErrors()) {
+            throw new IllegalArgumentException(errors.formatErrors());
+        }
+    }
+
+    private TotalComponent toTotalComponent(SensorMetadata metadata, int index, Errors errors) {
+        final var cm = metadata.metadataFor(index);
+        final var name = cm.name();
+        if (!cm.isWattCommensurable()) {
+            errors.addError("Component " + name
+                    + " is not commensurate with a power measure. It needs to be expressible in Watts.");
+        }
+
+        final var factor = SensorUnit.of(cm.unit()).getUnit().factor();
+        return new TotalComponent(name, index, factor);
+    }
+
+    public double total() {
+        return accumulatedTotal;
+    }
+
+    public double minMeasuredTotal() {
+        return minTotal;
+    }
+
+    public double maxMeasuredTotal() {
+        return maxTotal;
+    }
+
+    private record TotalComponent(String name, int index, double factor) {
+    }
+
+    @Override
+    public String name() {
+        return name;
+    }
+
+    @Override
+    public String output() {
+        return MeasureProcessor.super.output();
+    }
+
+    @Override
+    public void recordMeasure(double[] measure, long timestamp) {
+        final double recordedTotal = formula.apply(measure);
+        accumulatedTotal += recordedTotal;
+        if (recordedTotal < minTotal) {
+            minTotal = recordedTotal;
+        }
+        if (recordedTotal > maxTotal) {
+            maxTotal = recordedTotal;
+        }
+    }
+}