YamlRollingTimeWindowMeanEnricher extends AbstractTransformer<T,Double> { public ConfidenceQualifiedNumber getAverage(long fromTime, long graceAllowed) { if (timestamps.isEmpty()) { return lastAverage = new ConfidenceQualifiedNumber(lastAverage.value, 0.0d); } long firstTimestamp = -1; Iterator<Long> ti = timestamps.iterator(); while (ti.hasNext()) { firstTimestamp = ti.next(); if (firstTimestamp>0) break; } if (firstTimestamp<=0) { return lastAverage = new ConfidenceQualifiedNumber(values.get(values.size()-1).doubleValue(), 0.0d); } long lastTimestamp = timestamps.get(timestamps.size()-1); long now = fromTime; if (lastTimestamp > fromTime - graceAllowed) { now = lastTimestamp; } pruneValues(now); Duration timePeriod = getConfig(WINDOW_DURATION); long windowStart = Math.max(now-timePeriod.toMilliseconds(), firstTimestamp); long windowEnd = Math.max(now-timePeriod.toMilliseconds(), lastTimestamp); Double confidence = ((double)(windowEnd - windowStart)) / timePeriod.toMilliseconds(); if (confidence <= 0.0000001d) { double lastValue = values.get(values.size()-1).doubleValue(); return lastAverage = new ConfidenceQualifiedNumber(lastValue, 0.0d); } long start = windowStart; long end; double weightedAverage = 0.0d; Iterator<T> valuesIter = values.iterator(); Iterator<Long> timestampsIter = timestamps.iterator(); while (valuesIter.hasNext()) { Number val = valuesIter.next(); Long timestamp = timestampsIter.next(); if (val!=null && timestamp >= start) { end = timestamp; weightedAverage += ((end - start) / (confidence * timePeriod.toMilliseconds())) * val.doubleValue(); start = timestamp; } } return lastAverage = new ConfidenceQualifiedNumber(weightedAverage, confidence); }  @Override void setEntity(EntityLocal entity); ConfidenceQualifiedNumber getAverage(long fromTime, long graceAllowed); static ConfigKey<Duration> WINDOW_DURATION; static ConfigKey<Double> CONFIDENCE_REQUIRED_TO_PUBLISH; }
@Test public void testDefaultAverageWhenEmpty() { ConfidenceQualifiedNumber average = averager.getAverage(0, 0); assertEquals(average.value, 0d); assertEquals(average.confidence, 0.0d); }
@Test public void testNoRecentValuesAverage() { averager.onEvent(newDeltaSensorEvent(10, 0)); average = averager.getAverage(timePeriod.toMilliseconds()+1000, 0); assertEquals(average.value, 10d); assertEquals(average.confidence, 0d); }
@Test public void testNoRecentValuesUsesLastForAverage() { averager.onEvent(newDeltaSensorEvent(10, 0)); averager.onEvent(newDeltaSensorEvent(20, 10)); average = averager.getAverage(timePeriod.toMilliseconds()+1000, 0); assertEquals(average.value, 20d); assertEquals(average.confidence, 0d); }
@Test public void testSingleValueTimeAverage() { averager.onEvent(newDeltaSensorEvent(10, 1000)); average = averager.getAverage(1000, 0); assertEquals(average.confidence, 0d); }
@Test public void testTwoValueAverageForPeriod() { averager.onEvent(newDeltaSensorEvent(10, 1000)); averager.onEvent(newDeltaSensorEvent(10, 2000)); average = averager.getAverage(2000, 0); assertEquals(average.value, 10 /1d); assertEquals(average.confidence, 1d); }
@Test public void testMonospacedAverage() { averager.onEvent(newDeltaSensorEvent(10, 1000)); averager.onEvent(newDeltaSensorEvent(20, 1250)); averager.onEvent(newDeltaSensorEvent(30, 1500)); averager.onEvent(newDeltaSensorEvent(40, 1750)); averager.onEvent(newDeltaSensorEvent(50, 2000)); average = averager.getAverage(2000, 0); assertEquals(average.value, (20+30+40+50)/4d); assertEquals(average.confidence, 1d); }
@Test public void testWeightedAverage() { averager.onEvent(newDeltaSensorEvent(10, 1000)); averager.onEvent(newDeltaSensorEvent(20, 1100)); averager.onEvent(newDeltaSensorEvent(30, 1300)); averager.onEvent(newDeltaSensorEvent(40, 1600)); averager.onEvent(newDeltaSensorEvent(50, 2000)); average = averager.getAverage(2000, 0); assertEquals(average.value, (20*0.1d)+(30*0.2d)+(40*0.3d)+(50*0.4d)); assertEquals(average.confidence, 1d); }
@Test public void testConfidenceDecay() { averager.onEvent(newDeltaSensorEvent(10, 1000)); averager.onEvent(newDeltaSensorEvent(20, 1250)); averager.onEvent(newDeltaSensorEvent(30, 1500)); averager.onEvent(newDeltaSensorEvent(40, 1750)); averager.onEvent(newDeltaSensorEvent(50, 2000)); average = averager.getAverage(2250, 0); assertEquals(average.value, (30+40+50)/3d); assertEquals(average.confidence, 0.75d); average = averager.getAverage(2500, 0); assertEquals(average.value, (40+50)/2d); assertEquals(average.confidence, 0.5d); average = averager.getAverage(2750, 0); assertEquals(average.value, 50d); assertEquals(average.confidence, 0.25d); average = averager.getAverage(3000, 0); assertEquals(average.value, 50d); assertEquals(average.confidence, 0d); }