pulsecounter.ino

// PULSE_PIN, LED_BUILTIN
// WIFI_SSID, WIFI_PASSWORD
// MQTT_SERVER, MQTT_PORT
// MQTT_META_TOPIC, MQTT_TOTAL_TOPIC, MQTT_LASTMIN_TOPIC
#include "config.h"

#include <EspMQTTClient.h>

EspMQTTClient client(
  WIFI_SSID, WIFI_PASSWORD,
  MQTT_SERVER,
  NULL, NULL,
  "Orno", 1883
);

bool pulse_count_retrieved = false;
long prev_pulses_msg = 0; // millis
long prev_aggregate_msg = 0; // millis
const long retrieval_timeout = 10 * 1000; // 10 seconds
const long pulse_msg_interval = 1 * 1000; // 1 second
const long aggregate_msg_interval = 60 * 1000; // 1 minute

char mqtt_msg[sizeof(long) + 1];

// marking as volatile as they're written to in the interrupt handler,
// so an overly clever compiler might falsely get the idea to make
// them const
volatile bool publish_pulses_msg = false;
volatile long pulse_count = 0;

long prev_pulse_count = 0;
long aggregate_pulse_count = 0;

void onConnectionEstablished() {
  digitalWrite(LED_BUILTIN, HIGH);

  client.subscribe(MQTT_TOTAL_TOPIC, [] (const String & old_pulse_count) {
    if (!pulse_count_retrieved) {
      Serial.print("Old pulse count: ");
      Serial.println(old_pulse_count);
      // toInt() might just as well be called toLong()
      prev_pulse_count = pulse_count = old_pulse_count.toInt();
      elapsedkWhFromPulseCount();
      setPulseCountRetrieved();
    }
  });

  client.publish(MQTT_META_TOPIC, "connected", true /* retain */);
}

void setup()
{
  Serial.begin(115200);

  pinMode(LED_BUILTIN, OUTPUT);
  pinMode(PULSE_PIN, INPUT);

  // client.enableDebuggingMessages();
  // client.enableHTTPWebUpdater("admin", "orno", "/firmware");
  
  client.enableLastWillMessage(MQTT_META_TOPIC, "disconnected", true /* retain */);
}

void setPulseCountRetrieved() {
  pulse_count_retrieved = true;
  attachInterrupt(digitalPinToInterrupt(PULSE_PIN), onPulse, FALLING);
}

void pubElapsedTotalMsg() {
  digitalWrite(LED_BUILTIN, LOW);
  ltoa(pulse_count, mqtt_msg, 10 /* base, not size! */);
  client.publish(MQTT_TOTAL_TOPIC, mqtt_msg, true /* retain */);
  delay(10);
  digitalWrite(LED_BUILTIN, HIGH);
}

void pubLastMinMsg() {
  ltoa(aggregate_pulse_count, mqtt_msg, 10 /* base, not size! */);
  client.publish(MQTT_LASTMIN_TOPIC, mqtt_msg, false /* retain */);
}

void elapsedkWhFromPulseCount() {
  double elapsed_kWh = (1.0 * pulse_count / (1 * 1000));
  Serial.print("Elapsed total: ");
  Serial.print(elapsed_kWh, 3);
  Serial.println("kWh");
}

void aggregatekWhFromPulseCount() {
  double aggregate_kWh = (1.0 * aggregate_pulse_count / (1 * 1000));
  Serial.print("Last minute: ");
  Serial.print(aggregate_kWh, 3);
  Serial.println("kWh");
}

void loop() {
  client.loop();

  if (!pulse_count_retrieved && millis() >= retrieval_timeout) {
    setPulseCountRetrieved();
  }

  if (millis() >= prev_pulses_msg + pulse_msg_interval) {
    if (publish_pulses_msg) {
      elapsedkWhFromPulseCount();
      pubElapsedTotalMsg();
      publish_pulses_msg = false;
    }    
    prev_pulses_msg = millis();
  }

  if (millis() >= prev_aggregate_msg + aggregate_msg_interval) {
    aggregate_pulse_count = pulse_count - prev_pulse_count;
    aggregatekWhFromPulseCount();
    pubLastMinMsg();
    prev_pulse_count = pulse_count;
    prev_aggregate_msg = millis();
  }
}

void onPulse()
{
  ++pulse_count;
  publish_pulses_msg = true;
}