A shelly script to publish the state of a (shelly addon or uni ) digital input that counts them as impuls signals

impuls_sensor_from_digital_input_shelly_script.js

// Edit here =>
const CONFIG = {
  InputId: 100,
  Offset: 101189, // Initial count
  Factor: 0.01, // total = count * factor
  UnitOfMeasurement: 'm³',
  MinimalImpulseDuration: 2000, // ms
  UnpublishedCountThreshold: 2,
  DefaultPublishInterval: 60000, // ms
  MqttTopicNamespace: undefined, // Defaults to MqttConfig.topic_prefix or DeviceInfo.id
  MqttTopicSuffix: '/status/input:100:impulse',
  Debug: true,
};

// <= Stop edit here
const PublishCurrentState = (function() {
  let lastPublished = null;
  let lastPayload = undefined;
  let topic = undefined;

  return function() {
    if (!topic) topic = CONFIG.MqttTopicNamespace + CONFIG.MqttTopicSuffix;

    const now = Date.now();

    // Count differs initial value
    if (GLOBAL.ImpulseCounter.current() === CONFIG.Offset) {
      GLOBAL.Logger.debug('Current state will not published: Wait for first impulse');
      return;
    }

    const stateHasChanged = !lastPayload || (lastPayload.total !== GLOBAL.ImpulseCounter.total()) || (GLOBAL.ImpulseCounter.current() - lastPayload.current <= 0);

    if (stateHasChanged) {
      GLOBAL.Logger.debug('State has changed. Trigger impulse event');
      Shelly.emitEvent('impulse', { input_id: CONFIG.InputId, state: GLOBAL.ImpulseCounter.total() });
    }


    if (MQTT.isConnected() === false) {
      GLOBAL.Logger.debug('Current state cannot be published : MQTT is not connected!');
      return;
    }

    const publishIntervalNotExceeded = lastPublished && (now - lastPublished) < CONFIG.DefaultPublishInterval;
    const publishThresholdNotReached = lastPayload && (GLOBAL.ImpulseCounter.current() - lastPayload.current) < CONFIG.UnpublishedCountThreshold;

    if (publishIntervalNotExceeded && publishThresholdNotReached) {
      GLOBAL.Logger.debug('Current state will not published: Throttling according to configuration.');
      return;
    }

    const payload = lastPayload = {
      current: GLOBAL.ImpulseCounter.current(),
      total: GLOBAL.ImpulseCounter.total(),
      factor: CONFIG.Factor,
      offset: CONFIG.Offset,
      unit: CONFIG.UnitOfMeasurement,
      lastPublished: lastPublished,
      ts: now,
    };

    if (MQTT.publish(topic, JSON.stringify(payload), 1, false)) {
      lastPublished = now;
      GLOBAL.Logger.debug('Current state was published');
    }
  };
})();

const ImpulseCounterFactory = function(localCount) {
  let count = localCount || CONFIG.Offset || 0;
  let total = 0;

  const factorTotal = function() {
    return total = count * CONFIG.Factor;
  };

  // Initialize total
  factorTotal();

  return {
    current: function() {
      return count;
    },
    total: function() {
      return total;
    },
    Increase: function() {
      count++;
      factorTotal();

      UpdateLocalState();
      PublishCurrentState();
    },
  };
};

const TeardownBin = (function() {
  let pipeline = [];
  return {
    handler: function() {
      GLOBAL.Logger.debug('Execute teardown pipeline');
      for (let operator of pipeline) {
        operator();
      }
    },
    storage: {
      add: function(operator) {
        pipeline.push(operator);
      },
    },
  };
})();

const GLOBAL = {
  IsBootstrapped: false,
  DeviceInfo: Shelly.getDeviceInfo(),
  MqttConfig: {}, // Available after bootstrap
  ImpulseCounter: undefined, // Available after bootstrap
  TeardownPipeline: TeardownBin.storage,
  KvsStateKey: 'impulse_state',
  LastEvent: null,
  InitialInputState: null,
  Logger: {
    debug: function(msg) {
      if (!CONFIG.Debug) {
        return;
      }
      print(msg);
    },
    warn: function(msg) {
      print('!! ' + msg);
    },
    err: function(msg) {
      print('!!!! ' + msg);
    },
  },
};

function GetLocalState(cb, actualState) {
  Shelly.call('KVS.Get', { key: GLOBAL.KvsStateKey }, function(result, errCode, errMsg, _actualState) {
    if ((errCode || errMsg) && errCode !== -105 /*Not found*/) {
      console.error('Error while receiving value from KVS: [' + (errCode || 0) + '] ' + (errMsg || 'Unexpected error') + ' (Key: ' + GLOBAL.KvsStateKey + ')');
      return cb(null, true, _actualState);
    }

    if (errCode === -105) {
      return cb(undefined, false, _actualState);
    }

    const localStateValue = typeof result.value === 'string' ? JSON.parse(result.value) : result.value;

    const localState = {
      current: localStateValue.current,
      total: localStateValue.total,
      factor: localStateValue.factor,
      offset: localStateValue.offset,
    };

    cb(localState, false, _actualState);
  }, actualState);
}

function UpdateLocalState(cb) {
  if (typeof cb !== 'function') cb = function() {
  };
  const kvsData = {
    key: GLOBAL.KvsStateKey,
    value: JSON.stringify({
      current: GLOBAL.ImpulseCounter.current(),
      total: GLOBAL.ImpulseCounter.total(),
      factor: CONFIG.Factor,
      offset: CONFIG.Offset,
    }),
  };

  Shelly.call('KVS.Set', kvsData, function(result, errCode, errMsg) {
    if (errCode || errMsg) {
      console.error('Error while storing state in KVS: [' + (errCode || 0) + '] ' + (errMsg || 'Unexpected error') + ' (Key: ' + kvsData.key + '; Data: ' + kvsData.value + ')');
      return cb(null, true);
    }

    GLOBAL.Logger.debug('Current state has been stored to KVS');

    cb(result, false);
  });
}

function ImpulseEventHandler(event) {
  if (event.component !== ('input:' + CONFIG.InputId) || !event.info || typeof event.info.state !== 'boolean') {
    return;
  }

  GLOBAL.Logger.debug('Handle impulse event with state ' + event.info.state);

  const currentState = event.info.state;

  if (!GLOBAL.LastEvent && currentState === false) {
    // As we cannot determine the impulse as valid:
    //  ~~ No previous state and impulse might just ended with this event
    // we need to skip the count here
    return;
  }

  // When impulse incoming we prepare it for the event that end this impulse
  if (currentState === true) {
    GLOBAL.LastEvent = event;
    return;
  }

  const lastEventTs = GLOBAL.LastEvent.now * 1000;
  const currentEventTs = event.now * 1000;

  // Invalid impulse length have to be ignored
  if (Math.abs(currentEventTs - lastEventTs) < CONFIG.MinimalImpulseDuration) {
    GLOBAL.Logger.err('!!! Drop impulse due to invalid length (' + Math.round(Math.abs(currentEventTs - lastEventTs)) + 'ms)!');
    GLOBAL.LastEvent = null;
    return;
  }

  GLOBAL.Logger.debug('Impulse detected (Length: ' + Math.round(Math.abs(currentEventTs - lastEventTs)) + 'ms). Incrementing...');

  GLOBAL.ImpulseCounter.Increase();
}

function EventHandler(event) {
  if (event.info && event.info.event === 'scheduled_restart') {
    GLOBAL.Logger.debug('Device is about to restart');
    return TeardownBin.handler();
  }

  ImpulseEventHandler(event);
}

function setUpEnvironment(callback) {
  if (GLOBAL.IsBootstrapped) {
    return callback();
  }

  GLOBAL.Logger.debug('Set up environment');

  GLOBAL.TeardownPipeline.add(UpdateLocalState);
  GLOBAL.TeardownPipeline.add(function() {
    GLOBAL.Logger.debug('Pushing the last will: Good bye cruel world!');
    PublishCurrentState();
  });

  GLOBAL.KvsStateKey = 'input_' + CONFIG.InputId + '_' + GLOBAL.KvsStateKey;

  Shelly.call('Mqtt.GetConfig', { id: 1 }, function(mqttConfig) {
    GLOBAL.MqttConfig = mqttConfig;

    if (typeof CONFIG.MqttTopicNamespace !== 'string' || CONFIG.MqttTopicNamespace.length < 1) {
      CONFIG.MqttTopicNamespace = mqttConfig.topic_prefix || GLOBAL.DeviceInfo.id;
    }

    GetLocalState(function(localState) {
      let localCount = undefined;

      if (!localState) {
        GLOBAL.Logger.debug('Starting with a fresh state.');
      } else {
        let localAdd = 0;

        if (!isNaN(localState.offset) && localState.offset !== CONFIG.Offset) {
          if (localState.offset < CONFIG.Offset) {
            localAdd = Math.abs(CONFIG.Offset - localState.offset);
          }

          CONFIG.Offset = Math.max(localState.offset, CONFIG.Offset, 0);
          GLOBAL.Logger.warn('Config "Offset" has been changed. Using higher value: ' + CONFIG.Offset);
        }

        if (!isNaN(localState.factor) && localState.factor !== CONFIG.Factor) {
          GLOBAL.Logger.warn('Config "Factor" has been changed. No change on existing state was made. New factor: ' + CONFIG.Factor);
        }

        if (!isNaN(localState.current) && localState.current > CONFIG.Offset) {
          localCount = localState.current + localAdd;

          if (localAdd > 0) {
            GLOBAL.Logger.warn('Local value is adjusted to new offset: Increment "' + localAdd + '" to current value "' + localState.current + '". New value is "' + localCount + '"');
          }
        }
      }

      GLOBAL.ImpulseCounter = ImpulseCounterFactory(localCount);
      GLOBAL.LastEvent = {
        now: Date.now() - 500,
        info: {
          state: GLOBAL.InitialInputState,
        },
      };

      GLOBAL.Logger.debug('Initial local state - Count: ' + GLOBAL.ImpulseCounter.current() + ', Total: ' + GLOBAL.ImpulseCounter.total());

      callback();
    });
  });
}

function checkPrerequisitions(cb) {
  GLOBAL.Logger.debug('Check prerequisitions');

  Shelly.call('Input.GetStatus', { id: CONFIG.InputId }, function(result, errCode, errMsg) {
    if (errCode || errMsg) {
      GLOBAL.Logger.err('Cannot find Input (Id: ' + CONFIG.InputId + ')');
      return cb(false);
    }

    GLOBAL.InitialInputState = result.state;

    if (MQTT.isConnected() === false) {
      if (GLOBAL.MqttConfig.enable) {
        GLOBAL.Logger.warn('MQTT is not connected. The script will try to publish when connection is established.');
      } else {
        GLOBAL.Logger.err('MQTT is not enabled. As the impulse count cannot published on this device.');
//                return cb(false);
      }
    }

    GetLocalState(function(localState, hasError) {
      if (hasError) {
        GLOBAL.Logger.err('KVS cannot be accessed. This means the state count cannot persisted, which results in a state reset when device reboots.');
        return cb(false);
      }
    });

    return cb(true);
  });
}

function main() {
  GLOBAL.Logger.debug('Starting Impulse observation');

  const eventHandle = Shelly.addEventHandler(EventHandler);
  const intervalHandle = Timer.set(CONFIG.DefaultPublishInterval, true, PublishCurrentState);

  GLOBAL.TeardownPipeline.add(function() {
    GLOBAL.Logger.debug('Closing open handlers');
    Shelly.removeEventHandler(eventHandle);
    intervalHandle.clear();
  });
}

function bootstrap() {
  if (GLOBAL.IsBootstrapped) {
    return;
  }
  GLOBAL.Logger.debug('Bootstrap Impulse script');

  checkPrerequisitions(function(checkResult) {
    if (checkResult === false) {
      return;
    }

    setUpEnvironment(function() {
      GLOBAL.IsBootstrapped = true;

      UpdateLocalState();

      main();
    });
  });
}

bootstrap();