Introduce sustained counter profile.

This profile allows to build accumulated counter from values which gets reset on regular basis.
Most common inputs are daily yield counters as well as values which gets flushed during power cycles.

Signed-off-by: Łukasz Dywicki <luke@code-house.org>

bundles/org.connectorio.addons.profile.counter/src/main/java/org/connectorio/addons/profile/counter/internal/BaseCounterProfile.java

@@ -31,12 +31,18 @@
 public abstract class BaseCounterProfile implements StateProfile {
 
   protected final Logger logger = LoggerFactory.getLogger(BaseCounterProfile.class);
+  private final boolean lazyInitialization;
   protected final ProfileCallback callback;
   protected final UninitializedBehavior uninitializedBehavior;
   protected final ProfileContext context;
   protected Type last;
 
   protected BaseCounterProfile(ProfileCallback callback, ProfileContext context, LinkedItemStateRetriever linkedItemStateRetriever) {
+    this(false, callback, context, linkedItemStateRetriever);
+  }
+
+  protected BaseCounterProfile(boolean lazyInitialization, ProfileCallback callback, ProfileContext context, LinkedItemStateRetriever linkedItemStateRetriever) {
+    this.lazyInitialization = lazyInitialization;
     this.callback = callback;
     this.uninitializedBehavior = UninitializedBehavior.parse(context.getConfiguration().get("uninitializedBehavior"));
     this.context = context;
@@ -83,8 +89,8 @@ public void accept(T t) {
   }
 
   private void handleReading(Type val, boolean incoming) {
-    logger.trace("Verify reading {} vs {}. Value received from handler: {}", val, last, incoming);
-    if (last == null) {
+    logger.trace("Verify reading {} vs {}. Value received from {}", val, last, incoming ? "handler" : "item");
+    if (!lazyInitialization && last == null) {
       if (uninitializedBehavior == UninitializedBehavior.RESTORE_FROM_ITEM) {
         if (!incoming && val instanceof Command) {
           last = val;
@@ -126,4 +132,13 @@ public static UninitializedBehavior parse(Object behavior) {
       return RESTORE_FROM_ITEM;
     }
   }
+
+  static <T> boolean isLargerOrEqual(Comparable<T> current, T previous) {
+    return current.compareTo(previous) >= 0;
+  }
+
+  static <T> boolean isSmaller(Comparable<T> current, T previous) {
+    return current.compareTo(previous) < 0;
+  }
+
 }

bundles/org.connectorio.addons.profile.counter/src/main/java/org/connectorio/addons/profile/counter/internal/CounterProfileFactory.java

@@ -53,17 +53,22 @@ public Profile createProfile(ProfileTypeUID profileTypeUID, ProfileCallback call
     if (CounterProfiles.PULSE_COUNTER.equals(profileTypeUID)) {
       return new PulseCounterProfile(callback, profileContext, linkedItemStateRetriever);
     }
+    if (CounterProfiles.SUSTAINED_COUNTER.equals(profileTypeUID)) {
+      return new SustainedCounterProfile(callback, profileContext, linkedItemStateRetriever);
+    }
     return null;
   }
 
   @Override
   public Collection<ProfileTypeUID> getSupportedProfileTypeUIDs() {
-    return Arrays.asList(CounterProfiles.LIMIT_COUNTER_TOP, CounterProfiles.LIMIT_COUNTER_BOTTOM, CounterProfiles.PULSE_COUNTER);
+    return Arrays.asList(CounterProfiles.LIMIT_COUNTER_TOP, CounterProfiles.LIMIT_COUNTER_BOTTOM,
+      CounterProfiles.PULSE_COUNTER, CounterProfiles.SUSTAINED_COUNTER);
   }
 
   @Override
   public Collection<ProfileType> getProfileTypes(Locale locale) {
-    return Arrays.asList(CounterProfiles.LIMIT_COUNTER_TOP_PROFILE_TYPE, CounterProfiles.LIMIT_COUNTER_BOTTOM_PROFILE_TYPE, CounterProfiles.PULSE_PROFILE_TYPE);
+    return Arrays.asList(CounterProfiles.LIMIT_COUNTER_TOP_PROFILE_TYPE, CounterProfiles.LIMIT_COUNTER_BOTTOM_PROFILE_TYPE,
+      CounterProfiles.PULSE_PROFILE_TYPE, CounterProfiles.SUSTAINED_COUNTER_PROFILE_TYPE);
   }
 
 }

bundles/org.connectorio.addons.profile.counter/src/main/java/org/connectorio/addons/profile/counter/internal/CounterProfiles.java

@@ -36,9 +36,14 @@ public interface CounterProfiles {
     .build();
 
   ProfileTypeUID PULSE_COUNTER = new ProfileTypeUID("connectorio", "pulse-counter");
-  StateProfileType PULSE_PROFILE_TYPE = ProfileTypeBuilder.newState(LIMIT_COUNTER_TOP, "Pulse (tick) counter.")
+  StateProfileType PULSE_PROFILE_TYPE = ProfileTypeBuilder.newState(LIMIT_COUNTER_TOP, "Pulse (tick) counter")
     .withSupportedItemTypes("Number")
     .withSupportedItemTypesOfChannel("Switch", "Contact")
     .build();
 
+  ProfileTypeUID SUSTAINED_COUNTER = new ProfileTypeUID("connectorio", "sustained-counter");
+  StateProfileType SUSTAINED_COUNTER_PROFILE_TYPE = ProfileTypeBuilder.newState(SUSTAINED_COUNTER, "Sustained counter")
+    .withSupportedItemTypes("Number")
+    .withSupportedItemTypesOfChannel("Number")
+    .build();
 }

bundles/org.connectorio.addons.profile.counter/src/main/java/org/connectorio/addons/profile/counter/internal/LimitCounterBottomProfile.java

@@ -63,7 +63,7 @@ protected void handleReading(Type current, Type previous, boolean incoming) {
 
   private void compare(DecimalType current, DecimalType previous, Consumer<DecimalType> consumer) {
     logger.trace("Verify value {} is larger than {}", current, previous);
-    if (current.compareTo(previous) >= 0) {
+    if (isLargerOrEqual(current, previous)) {
       consumer.accept(current);
       return;
     }
@@ -73,7 +73,7 @@ private void compare(DecimalType current, DecimalType previous, Consumer<Decimal
   @SuppressWarnings({"unchecked", "rawtypes"})
   private void compare(QuantityType current, QuantityType previous, Consumer<QuantityType> consumer) {
     logger.trace("Verify value {} is larger than {}", current, previous);
-    if (current.compareTo(previous) >= 0) {
+    if (isLargerOrEqual(current, previous)) {
       consumer.accept(current);
       return;
     }

bundles/org.connectorio.addons.profile.counter/src/main/java/org/connectorio/addons/profile/counter/internal/LimitCounterTopProfile.java

@@ -35,7 +35,7 @@
 
 /**
  * A profile which makes sure that item receives only increasing values. It also makes sure that
- * received value is not to great.
+ * received value is not too great.
  *
  * @author Lukasz Dywicki - Initial contribution
  */
@@ -79,7 +79,7 @@ private void compare(DecimalType val, DecimalType last, Consumer<DecimalType> va
     BigDecimal lastReading = last.toBigDecimal();
     BigDecimal maxIncrease = lastReading.multiply(anomaly).divide(_100, RoundingMode.HALF_UP);
     logger.trace("Verify value {} is smaller than {} + {} ({})%", val, last, maxIncrease, anomaly);
-    if (currentReading.compareTo(lastReading.add(maxIncrease)) <= 0) {
+    if (isSmaller(currentReading, lastReading.add(maxIncrease))) {
       value.accept(val);
       return;
     }
@@ -95,7 +95,7 @@ private <T extends Quantity<T>> void compare(QuantityType<T> val, QuantityType<?
     }
     BigDecimal maxIncrease = lastReading.multiply(anomaly).divide(_100, RoundingMode.HALF_UP);
     logger.trace("Verify value {} is smaller than {} + {} ({})%", val, last, maxIncrease, anomaly);
-    if (currentReading.compareTo(lastReading.add(maxIncrease)) <= 0) {
+    if (isSmaller(currentReading, lastReading.add(maxIncrease))) {
       value.accept(val);
       return;
     }

bundles/org.connectorio.addons.profile.counter/src/main/java/org/connectorio/addons/profile/counter/internal/SustainedCounterProfile.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright (C) 2024-2024 ConnectorIO Sp. z o.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.connectorio.addons.profile.counter.internal;
+
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import java.util.function.BinaryOperator;
+import java.util.function.Consumer;
+import org.connectorio.addons.profile.counter.internal.state.LinkedItemStateRetriever;
+import org.openhab.core.library.types.DecimalType;
+import org.openhab.core.library.types.QuantityType;
+import org.openhab.core.thing.profiles.ProfileCallback;
+import org.openhab.core.thing.profiles.ProfileContext;
+import org.openhab.core.thing.profiles.ProfileTypeUID;
+import org.openhab.core.types.State;
+import org.openhab.core.types.Type;
+
+/**
+ * Little utility profile which tracks internally progression of values reported from thing handler.
+ * When value decreases it rests its own state to this value and sends it as delta. If value increases
+ * it calculates delta from base value registered earlier.
+ */
+class SustainedCounterProfile extends BaseCounterProfile {
+
+  private AtomicReference<State> previousValue = new AtomicReference<>();
+
+  SustainedCounterProfile(ProfileCallback callback, ProfileContext context, LinkedItemStateRetriever linkedItemStateRetriever) {
+    super(true, callback, context, linkedItemStateRetriever);
+  }
+
+  @Override
+  public ProfileTypeUID getProfileTypeUID() {
+    return CounterProfiles.PULSE_COUNTER;
+  }
+
+  @Override
+  @SuppressWarnings({"unchecked", "rawtypes"})
+  protected void handleReading(Type current, Type previous, boolean incoming) {
+    if (incoming) { // incoming, handler to item
+      if (current instanceof DecimalType) {
+        this.update((AtomicReference) previousValue, (DecimalType) current, (DecimalType) last, update(callback::sendUpdate),
+          (left, right) -> new DecimalType(left.toBigDecimal().subtract(right.toBigDecimal())),
+          (left, right) -> new DecimalType(left.toBigDecimal().add(right.toBigDecimal()))
+        );
+      } else if (current instanceof QuantityType) {
+        this.update((AtomicReference) previousValue, (QuantityType) current, (QuantityType) last, update(callback::sendUpdate),
+          QuantityType::subtract,
+          QuantityType::add
+        );
+      }
+    }
+    // outgoing communication is not supported
+  }
+
+
+  private <T extends Comparable<T>> void update(AtomicReference<T> reference, T current, T previous, Consumer<T> consumer, BiFunction<T, T, T> diff, BiFunction<T, T, T> sum) {
+    if (previous == null) {
+      if (reference.compareAndSet(null, current)) {
+        consumer.accept(current);
+        return;
+      }
+    }
+
+    if (reference.compareAndSet(null, current)) {
+      consumer.accept(sum.apply(previous, current));
+    } else {
+      reference.accumulateAndGet(current, new BinaryOperator<T>() {
+        @Override
+        public T apply(T prev, T next) {
+          T delta = null;
+          if (isLargerOrEqual(next, prev)) {
+            delta = diff.apply(next, prev);
+          } else if (isSmaller(next, prev)) {
+            delta = next;
+          }
+          if (delta != null) {
+            consumer.accept(sum.apply(previous, delta));
+          }
+          return current;
+        }
+      });
+    }
+  }
+
+}

bundles/org.connectorio.addons.profile.counter/src/main/resources/OH-INF/config/sustained-counter.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ - Copyright (C) 2024-2024 ConnectorIO Sp. z o.o.
+ -
+ - Licensed under the Apache License, Version 2.0 (the "License");
+ - you may not use this file except in compliance with the License.
+ - You may obtain a copy of the License at
+ -
+ -     http://www.apache.org/licenses/LICENSE-2.0
+ -
+ - Unless required by applicable law or agreed to in writing, software
+ - distributed under the License is distributed on an "AS IS" BASIS,
+ - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ - See the License for the specific language governing permissions and
+ - limitations under the License.
+ -
+ - SPDX-License-Identifier: Apache-2.0
+ -->
+<config-descriptions
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="https://openhab.org/schemas/config-description/v1.0.0 https://openhab.org/schemas/config-description-1.0.0.xsd">
+
+  <config-description uri="profile:connectorio:pulse-counter">
+    <parameter name="uninitializedBehavior" type="text" required="true">
+      <label>Uninitialized behavior</label>
+      <description>
+        Defines behavior when first value arrive and what to do with it. There are two cases which needs to be considered:
+      </description>
+      <options>
+        <option value="RESTORE_FROM_ITEM">Require receiving of item state first</option>
+        <option value="RESTORE_FROM_PERSISTENCE">Restore last value from default persistence</option>
+        <option value="RESTORE_FROM_HANDLER">Require receiving of state from device.</option>
+      </options>
+      <default>RESTORE_FROM_ITEM</default>
+    </parameter>
+  </config-description>
+
+</config-descriptions>

bundles/org.connectorio.addons.profile.counter/src/test/java/org/connectorio/addons/profile/counter/internal/LimitCounterBottomProfileTest.java

@@ -53,13 +53,12 @@ class LimitCounterBottomProfileTest {
   @Test
   void checkDecimalValueFromItem() {
     HashMap<String, Object> cfgMap = new HashMap<>();
-    cfgMap.put("anomaly", "10");
     Configuration config = new Configuration(cfgMap);
 
     when(context.getConfiguration()).thenReturn(config);
 
     // default -> first state from item!
-    LimitCounterTopProfile profile = new LimitCounterTopProfile(callback, context, itemStateRetriever);
+    LimitCounterBottomProfile profile = new LimitCounterBottomProfile(callback, context, itemStateRetriever);
 
     // update from handler before item state is set -> no interactions
     profile.onStateUpdateFromHandler(new DecimalType(13.0));
@@ -72,23 +71,22 @@ void checkDecimalValueFromItem() {
     Mockito.verify(callback).sendUpdate(new DecimalType(10.1));
 
     profile.onStateUpdateFromHandler(new DecimalType(13.0));
-    Mockito.verifyNoMoreInteractions(callback);
+    Mockito.verify(callback).sendUpdate(new DecimalType(13.0));
 
     profile.onStateUpdateFromHandler(new DecimalType(11.0));
-    Mockito.verify(callback).sendUpdate(new DecimalType(11.0));
+    Mockito.verifyNoMoreInteractions(callback);
   }
 
   @Test
   void checkDecimalValueFromHandler() {
     HashMap<String, Object> cfgMap = new HashMap<>();
-    cfgMap.put("anomaly", "10");
     cfgMap.put("uninitializedBehavior", UninitializedBehavior.RESTORE_FROM_HANDLER.name());
     Configuration config = new Configuration(cfgMap);
 
     when(context.getConfiguration()).thenReturn(config);
 
     // default -> first state from item!
-    LimitCounterTopProfile profile = new LimitCounterTopProfile(callback, context, itemStateRetriever);
+    LimitCounterBottomProfile profile = new LimitCounterBottomProfile(callback, context, itemStateRetriever);
 
     // update from handler before item state is set -> no interactions
     profile.onStateUpdateFromItem(new DecimalType(13.0));
@@ -102,34 +100,35 @@ void checkDecimalValueFromHandler() {
   }
 
   @Test
-  @SuppressWarnings("null")
   void checkDecimalValueFromPersistence() {
     HashMap<String, Object> cfgMap = new HashMap<>();
-    cfgMap.put("anomaly", "10");
     cfgMap.put("uninitializedBehavior", UninitializedBehavior.RESTORE_FROM_PERSISTENCE.name());
     Configuration config = new Configuration(cfgMap);
 
     when(context.getConfiguration()).thenReturn(config);
+    when(itemStateRetriever.getItemName(callback)).thenReturn("foo");
+    when(itemStateRetriever.retrieveState("foo")).thenReturn(new DecimalType(10));
 
-    // we shall start with 10.0 retrieved from persistence
+    // we shall start with 10.0 retrieved from persistence, so we should accept values below 11
     LimitCounterBottomProfile profile = new LimitCounterBottomProfile(callback, context, itemStateRetriever);
 
     // update from item above accepted level
     profile.onStateUpdateFromItem(new DecimalType(13.0));
-    Mockito.verifyNoInteractions(callback);
+    Mockito.verify(callback).handleCommand(new DecimalType(13.0));
 
-    // anomaly call
+    // ??
     profile.onStateUpdateFromHandler(new DecimalType(13.0));
-    Mockito.verifyNoInteractions(callback);
+    Mockito.verify(callback).sendUpdate(new DecimalType(13.0));
+
     // accepted call
     profile.onStateUpdateFromHandler(new DecimalType(10.1));
-    Mockito.verifyNoInteractions(callback);
+
+    Mockito.verifyNoMoreInteractions(callback);
   }
 
   @Test
   void checkQuantityValueFromItem() {
     HashMap<String, Object> cfgMap = new HashMap<>();
-    cfgMap.put("anomaly", "10");
     Configuration config = new Configuration(cfgMap);
 
     when(context.getConfiguration()).thenReturn(config);

bundles/org.connectorio.addons.profile.counter/src/test/java/org/connectorio/addons/profile/counter/internal/LimitCounterTopProfileTest.java

@@ -104,18 +104,17 @@ void checkDecimalValueFromHandler() {
   }
 
   @Test
-  @SuppressWarnings("null")
   void checkDecimalValueFromPersistence() {
     HashMap<String, Object> cfgMap = new HashMap<>();
-    cfgMap.put("anomaly", "10");
+    cfgMap.put("anomaly", "10"); // max change is 10%
     cfgMap.put("uninitializedBehavior", UninitializedBehavior.RESTORE_FROM_PERSISTENCE.name());
     Configuration config = new Configuration(cfgMap);
 
     when(context.getConfiguration()).thenReturn(config);
     when(itemStateRetriever.getItemName(callback)).thenReturn("foo");
     when(itemStateRetriever.retrieveState("foo")).thenReturn(new DecimalType(10.0));
 
-    // we shall start with 10.0 retrieved from persistence
+    // we shall start with 10.0 retrieved from persistence, so we should only accept values below 11
     LimitCounterTopProfile profile = new LimitCounterTopProfile(callback, context, itemStateRetriever);
 
     // update from item above accepted level