Detect repeated whole-hour meter registers, and announce state over MQTT (#1247)

* Detect repeated whole-hour meter registers, and announce state over MQTT

Two coupled changes: the fix for #1119, and the MQTT announcements from #1128
that make the correction visible instead of silent.

Reject repeated whole-hour registers (#1119)

Some meters publish the previous whole hour's accumulated registers again at the
next whole hour: identical registers and an identical meter clock, while the
instantaneous values in the same payload are current. Confirmed on three Aidon
meters; the four captured payloads are the gh1119-1..4 fixtures.

Stored history was credited with a zero hour followed by a double hour, which
destroys the month's tariff peak until the month rolls over. The 2.3 workaround
in AmsData::apply() never took effect on history at all: it substituted the
integrated estimate into meterState, but handleDataSuccess passes the raw packet
to ds.update(), and meterState is never used to build history.

AmsData::isStaleCounter() now detects the repeat, meter-agnostically, from the
meter clock: the timestamp equals the last accepted List >= 3 timestamp, the
registers are unchanged, and the meter's own observed register cadence is at
least 30 minutes. It gives up after two consecutive rejections so a frozen clock
cannot suppress history indefinitely. The cadence is learned from the meter
rather than compared against the system clock, because adjustForKnownIssues()
applies a flat -3600 for Aidon that is only correct outside DST; learning the
step also keeps meters that publish registers every few seconds out of scope.

For the affected hour, handleDataSuccess stores the power-integrated estimate
once the device has been up for an hour, so the next hour's delta is measured
against the estimate and the two-hour total closes exactly on the meter's own
register. Before that there is no reliable baseline, so nothing is stored and
the next whole hour falls into AmsDataStorage's existing average branch, which
distributes the two-hour delta evenly across both hours. AmsDataStorage itself
is unchanged.

The Aidon-specific gate, the exact-double counter compare and the diff < 1.0
case are gone. That last one accepted the stale value whenever the hour's
estimate was below 1 kWh, which is most nights.

The four MQTT handlers withhold the repeated accumulated registers; the
instantaneous values in the same payload are current and still go out. The flag
describes the packet most recently applied and is cleared by the next sub-hour
packet, so the state snapshots the handlers compose in stateUpdate mode do not
inherit it for the rest of the hour.

Announce service state and events over MQTT (#1128)

A degraded service was only visible in the web UI and as red LED blinks. Nothing
reached MQTT, so a Home Assistant user could not be told that the reader stopped
receiving meter data, that an hourly reading had to be substituted, or that
price fetching was failing (#1120).

<topic>/services carries a retained state snapshot, built from the same
buildServicesJson() the web UI is served from so there is one vocabulary rather
than two. It is republished on change or once a minute. The detail fields are
omitted: the MQTT packet buffer is 256 bytes on ESP8266 and a meter model plus a
broker hostname plus an NTP server would not fit. buildServicesJson() gained a
withDetail flag, and its nine duplicated entry blocks were folded into one
helper so the two shapes cannot drift; the web payload is unchanged.

<topic>/event carries one message per transition, not retained:
hourly_data_substituted, han_no_data, han_undecodable, han_error, han_restored,
price_error and price_ok. The raw error code travels with the event rather than
being mapped to a category in firmware, since the codes are already documented
and translated for the UI. HAN events are rate limited to one per 30 s on top of
the debounce already in AmsData::getLastError().

<topic>/status was already the retained online/offline availability payload
backed by the last will, which is why the snapshot uses /services - and why
adding avty_t to the Home Assistant discovery template was a single line. Home
Assistant also gets a device_class:problem diagnostic binary sensor, which is
the one entity a notification automation can trigger on, plus a sensor for the
raw HAN error code.

Verified with six native tests over the gh1119 fixtures (pio test -e native, 23
cases) and builds for esp8266 and all six ESP32 targets. ESP8266 flash goes from
98.0% to 98.3% of the 1 MB budget.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Move the shared services JSON generator into AmsJsonGenerator

buildServicesJson() stopped being web-only when the MQTT announcements started
rendering from it, which left the MQTT path reaching through the web server for
state it does not own. It now lives with the other shared generators, alongside
hanState() and mqttHandlerState().

A ServiceStatusContext carries the services to report on, so both callers fill in
what they have rather than the generator depending on either. AmsWebServer keeps
computeServicesAggregate(), which is web-only, but builds it from the moved
helpers so the badge, the services array and the MQTT announcements cannot
disagree.

No behaviour change: the web payload is byte for byte what it was.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Drop the HAN error sensor and the duplicated top-level status fields

The services snapshot carried up/han/hanError beside the array, and a "HAN error"
sensor reading the raw code. han and hanError restated services[0], up restated
the /state payload, and the sensor restated the problem binary sensor with less
meaning attached.

Only "problem" stays outside the array. It is computed rather than duplicated: it
saves every subscriber from indexing into the array by position to find out
whether anything is wrong, and it is what the Home Assistant binary sensor
templates on. A four-service payload is now 126 bytes plus the topic, well inside
the 256 byte MQTT packet buffer on ESP8266.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Add a Home Assistant event entity, and move "problem" to the system payload

Three changes, one of them a correction.

Correct the ESP8266 packet buffer claim

The services payload was trimmed of its "d" detail fields on the belief that the
256 byte MQTTClient buffer on ESP8266 caps outbound payloads. It does not.
lwmqtt_encode_publish() writes only the fixed header and the topic into the write
buffer; lwmqtt_write_to_network() then streams the payload straight to the socket.
The 256 bytes cap *inbound* messages, which is what dropOverflow(true) is for.
That is also why the ~400 byte Home Assistant discovery configs have always
published fine from ESP8266.

So the withDetail flag is gone and MQTT gets the same array the web UI does,
detail included.

Move "problem" to the system payload

It was the only field outside the services array, which left the array topic
carrying a field that is not a service. It now rides along with the other
device-level health fields in the system payload, so <topic>/services is nothing
but the array. The Home Assistant binary sensor follows it to <topic>/state,
alongside the rssi/vcc/uptime sensors that already read from there. The JSON
payload format carries the same field in its own system payload.

Handlers reach the meter state through a new setMeterState(), matching the
existing setDataStorage() wiring. Cadence drops from at most every five seconds
to the 60 second system publish, comfortably inside the sensor's 300 s
expire_after.

Add the event entity

Home Assistant's MQTT event platform reads "event_type" out of the payload
natively and turns the remaining keys into attributes, so publishEvent() names
the field that way and no value_template is needed. The entity declares all seven
event types and lands under Diagnostic on the device page; the transitions are
now visible in Home Assistant rather than only to an mqtt trigger.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: gskjold

src/AmsData.cpp

@@ -41,6 +41,13 @@ void AmsData::apply(AmsData& other) {
             }
             counterEstimated = true;
         }
+
+        // The flag describes the packet most recently applied. Sub-hour traffic
+        // means the stale whole-hour repeat is no longer the newest thing we know,
+        // so clear it: otherwise the state snapshots that the MQTT handlers
+        // compose from this state would inherit it for the rest of the hour and
+        // withhold counters that are perfectly good.
+        this->counterStale = false;
     }
 
     this->lastUpdateMillis = other.getLastUpdateMillis();
@@ -70,21 +77,17 @@ void AmsData::apply(AmsData& other) {
             this->l3activeExportCounter = other.getL3ActiveExportCounter();
         case 3:
             this->meterTimestamp = other.getMeterTimestamp();
-            // Aidon tends to sometime send the same counter as last hour by accident
-            if(meterType == AmsTypeAidon && counterEstimated && lastKnownCounter == other.getActiveImportCounter()-other.getActiveExportCounter()) {
-                double diff = activeImportCounter - activeExportCounter - lastKnownCounter;
-                if(diff < 1.0) { // In case a very low value have been calculated, use the new values
-                    this->activeImportCounter = other.getActiveImportCounter();
-                    this->activeImportCounterTariff1 = other.getActiveImportCounterTariff1();
-                    this->activeImportCounterTariff2 = other.getActiveImportCounterTariff2();
-                    this->activeExportCounter = other.getActiveExportCounter();
-                    this->activeExportCounterTariff1 = other.getActiveExportCounterTariff1();
-                    this->activeExportCounterTariff2 = other.getActiveExportCounterTariff2();
-                    this->reactiveImportCounter = other.getReactiveImportCounter();
-                    this->reactiveExportCounter = other.getReactiveExportCounter();
-                    this->lastKnownCounter = activeImportCounter - activeExportCounter;
-                }
+            // Some meters repeat the previous whole hour's register snapshot at the
+            // next whole hour (#1119). Keep the integrated estimate and stay flagged
+            // as estimated, so the repeat is neither stored nor published as a new
+            // hourly reading. The flag is set by the pipeline entry point via
+            // isStaleCounter() before this point, and propagates through apply().
+            if(other.isCounterStale()) {
+                this->counterStale = true;
+                if(this->staleCounterCount < 0xFF) this->staleCounterCount++;
             } else {
+                this->counterStale = false;
+                this->staleCounterCount = 0;
                 this->activeImportCounter = other.getActiveImportCounter();
                 this->activeImportCounterTariff1 = other.getActiveImportCounterTariff1();
                 this->activeImportCounterTariff2 = other.getActiveImportCounterTariff2();
@@ -94,8 +97,14 @@ void AmsData::apply(AmsData& other) {
                 this->reactiveImportCounter = other.getReactiveImportCounter();
                 this->reactiveExportCounter = other.getReactiveExportCounter();
                 this->lastKnownCounter = activeImportCounter - activeExportCounter;
+                if(other.getMeterTimestamp() != 0) {
+                    if(this->lastAcceptedMeterTimestamp != 0 && other.getMeterTimestamp() > this->lastAcceptedMeterTimestamp) {
+                        this->lastAcceptedMeterTimestampStep = (int32_t) (other.getMeterTimestamp() - this->lastAcceptedMeterTimestamp);
+                    }
+                    this->lastAcceptedMeterTimestamp = other.getMeterTimestamp();
+                }
+                this->counterEstimated = false;
             }
-            this->counterEstimated = false;
         case 2:
             strncpy(this->listId, other.listId, sizeof(this->listId) - 1);
             strncpy(this->meterId, other.meterId, sizeof(this->meterId) - 1);
@@ -469,6 +478,39 @@ bool AmsData::isCounterEstimated() {
     return this->counterEstimated;
 }
 
+bool AmsData::isStaleCounter(AmsData& other) {
+    if(other.getListType() < 3) return false;
+
+    // Never suppress more than two consecutive readings. A meter with a frozen
+    // clock must not be able to blackhole history indefinitely.
+    if(this->staleCounterCount >= 2) return false;
+
+    time_t ts = other.getMeterTimestamp();
+    if(ts == 0 || this->lastAcceptedMeterTimestamp == 0) return false; // no clock to compare against
+    if(ts != this->lastAcceptedMeterTimestamp) return false;           // clock advanced, genuine new reading
+
+    // Only meters that publish their registers on a long cycle can exhibit this.
+    // A meter sending List 3 every few seconds repeats its clock between updates
+    // as a matter of course, so require an observed hourly cadence first. The
+    // step is learned from the meter itself rather than compared against the
+    // system clock: adjustForKnownIssues() gives some meters a fixed offset that
+    // is only correct outside DST, so absolute comparison is not reliable.
+    if(this->lastAcceptedMeterTimestampStep < 1800) return false;
+
+    // Clock repeated. Require the registers to be identical too - both signals
+    // are present in every captured case, so demanding both avoids false
+    // positives on meters with an unreliable clock.
+    return this->lastKnownCounter == other.getActiveImportCounter() - other.getActiveExportCounter();
+}
+
+bool AmsData::isCounterStale() {
+    return this->counterStale;
+}
+
+void AmsData::setCounterStale(bool stale) {
+    this->counterStale = stale;
+}
+
 bool AmsData::isL2currentMissing() {
     return this->l2currentMissing;
 }

src/AmsData.h

@@ -29,6 +29,13 @@ class AmsData {
     void apply(AmsData& other);
     void apply(const OBIS_code_t obis, double value, uint64_t millis64);
 
+    // True when other's List >= 3 accumulated registers are a verbatim repeat of
+    // the last set we accepted: same meter clock AND same counters. Some meters
+    // (confirmed on Aidon, #1119) publish the previous hour's register snapshot
+    // again at the next whole hour, which would otherwise store a zero hour
+    // followed by a double hour. Pure query - apply() does the bookkeeping.
+    bool isStaleCounter(AmsData& other);
+
     uint64_t getLastUpdateMillis();
 
     time_t getPackageTimestamp();
@@ -85,6 +92,13 @@ class AmsData {
     double getActiveExportCounterTariff2();
     double getReactiveExportCounter();
 
+    // Whether the most recently applied packet was a stale whole-hour repeat.
+    // Set from isStaleCounter() by the pipeline entry point, cleared again by the
+    // next sub-hour packet, and consulted by the MQTT handlers to withhold the
+    // repeated accumulated registers.
+    bool isCounterStale();
+    void setCounterStale(bool);
+
     bool isThreePhase();
     bool isTwoPhase();
     bool isCounterEstimated();
@@ -112,7 +126,11 @@ class AmsData {
     float powerFactor = 0, l1PowerFactor = 0, l2PowerFactor = 0, l3PowerFactor = 0;
     double activeImportCounter = 0, activeImportCounterTariff1 = 0, activeImportCounterTariff2 = 0, reactiveImportCounter = 0, activeExportCounter = 0, activeExportCounterTariff1 = 0, activeExportCounterTariff2 = 0, reactiveExportCounter = 0;
     double lastKnownCounter = 0;
+    time_t lastAcceptedMeterTimestamp = 0;
+    int32_t lastAcceptedMeterTimestampStep = 0;
+    uint8_t staleCounterCount = 0;
     bool threePhase = false, twoPhase = false, counterEstimated = false, l2currentMissing = false;;
+    bool counterStale = false;
 
     int8_t lastError = 0x00;
     uint8_t lastErrorCount = 0;

src/AmsJsonGenerator.cpp

@@ -8,6 +8,173 @@
 #include "AmsStorage.h"
 #include "LittleFS.h"
 #include "FirmwareVersion.h"
+#include "AmsData.h"
+#include "PriceService.h"
+#include "NtpStatus.h"
+#include "Uptime.h"
+#include "mqtt/AmsMqttHandler.h"
+#if defined(AMS_CLOUD)
+#include "cloud/CloudConnector.h"
+#endif
+#if defined(ZMART_CHARGE)
+#include "cloud/ZmartChargeCloudConnector.h"
+#endif
+
+// SNTP resyncs roughly hourly; flag the NTP service as degraded if no sync has
+// landed in this long, allowing a couple of missed cycles before warning.
+#define NTP_STALE_AFTER_SECONDS 10800
+
+uint8_t AmsJsonGenerator::hanState(AmsData* meterState) {
+    if(meterState == NULL) return 2;
+    uint64_t millis = millis64();
+    if(meterState->getLastError() != 0) return 3;
+    // State 0 means disabled, which the HAN port never is. Waiting for the first
+    // frame after boot is the connecting state.
+    if(meterState->getLastUpdateMillis() == 0 && millis < 30000) return 2;
+    if(millis - meterState->getLastUpdateMillis() < 15000) return 1;
+    if(millis - meterState->getLastUpdateMillis() < 30000) return 2;
+    return 3;
+}
+
+uint8_t AmsJsonGenerator::mqttHandlerState(AmsMqttHandler* h) {
+    if(h == NULL) return 2;
+    if(h->connected()) return 1;
+    return h->lastError() == 0 ? 2 : 3;
+}
+
+// Formats one entry of the services array. Kept in one place so the web payload
+// and the MQTT payload (#1128) cannot drift apart.
+static void appendServiceEntry(String& out, const char* key, uint8_t state, int16_t err, const char* detail, const char* name) {
+    char entry[320];
+    snprintf_P(entry, sizeof(entry), PSTR("{\"k\":\"%s\",\"s\":%d,\"e\":%d%s%s%s,\"d\":\"%s\"}"),
+        key, state, err,
+        name != NULL ? ",\"n\":\"" : "", name != NULL ? name : "", name != NULL ? "\"" : "",
+        detail == NULL ? "" : detail);
+    if(!out.isEmpty()) out += ",";
+    out += entry;
+}
+
+String AmsJsonGenerator::generateServicesJson(const ServiceStatusContext& ctx) {
+    String out = "";
+    if(ctx.config == NULL) return out;
+
+    {
+        String meterModel = ctx.meterState == NULL ? String("") : String(ctx.meterState->getMeterModel());
+        if(!meterModel.isEmpty())
+            meterModel.replace(F("\\"), F("\\\\"));
+        appendServiceEntry(out, "han", hanState(ctx.meterState),
+            ctx.meterState == NULL ? 0 : ctx.meterState->getLastError(), meterModel.c_str(), NULL);
+    }
+
+    MqttConfig mqttConfig;
+    bool haveMqttConfig = ctx.config->getMqttConfig(mqttConfig);
+    if(haveMqttConfig && strlen(mqttConfig.host) > 0) {
+        uint8_t s;
+        int16_t err = 0;
+        if(!ctx.mqttEnabled) {
+            s = 0;
+        } else {
+            s = mqttHandlerState(ctx.mqttHandler);
+            if(ctx.mqttHandler != NULL) err = (int16_t) ctx.mqttHandler->lastError();
+        }
+        appendServiceEntry(out, "mqtt", s, err, mqttConfig.host, NULL);
+    }
+
+    #if defined(CUSTOM_MQTT_HOST)
+    {
+        uint8_t s = mqttHandlerState(ctx.customMqttHandler);
+        int16_t err = ctx.customMqttHandler == NULL ? 0 : (int16_t) ctx.customMqttHandler->lastError();
+        #if defined(CUSTOM_MQTT_NAME)
+        appendServiceEntry(out, "mqtt_c", s, err, CUSTOM_MQTT_HOST, CUSTOM_MQTT_NAME);
+        #else
+        appendServiceEntry(out, "mqtt_c", s, err, CUSTOM_MQTT_HOST, NULL);
+        #endif
+    }
+    #endif
+
+    #if defined(ESP32) && defined(ENERGY_SPEEDOMETER_PASS)
+    {
+        SystemConfig sys;
+        ctx.config->getSystemConfig(sys);
+        if(sys.energyspeedometer == 7) {
+            uint8_t s = mqttHandlerState(ctx.energySpeedometer);
+            int16_t err = ctx.energySpeedometer == NULL ? 0 : (int16_t) ctx.energySpeedometer->lastError();
+            appendServiceEntry(out, "mqtt_es", s, err, "", NULL);
+        }
+    }
+    #endif
+
+    PriceServiceConfig priceCfg;
+    if(ctx.config->getPriceServiceConfig(priceCfg) && priceCfg.enabled && strlen(priceCfg.area) > 0) {
+        uint8_t s;
+        int16_t err = ctx.ps == NULL ? 0 : ctx.ps->getLastError();
+        if(ctx.ps == NULL) {
+            s = 2;
+        } else if(err != 0) {
+            s = 3;
+        } else if(ctx.ps->hasPrice()) {
+            s = 1;
+        } else {
+            s = 2;
+        }
+        appendServiceEntry(out, "price", s, err, priceCfg.area, NULL);
+    }
+
+    {
+        NtpConfig ntp;
+        if(ctx.config->getNtpConfig(ntp) && ntp.enable) {
+            const char* server = strlen(ntp.server) > 0 ? ntp.server : "pool.ntp.org";
+            // A set-but-stale clock (NTP stopped resyncing) silently corrupts
+            // day-boundary accounting, so flag staleness rather than only
+            // reporting whether the clock was ever set.
+            uint64_t lastSync = ntpLastSyncMillis();
+            uint8_t s;
+            if(lastSync == 0) {
+                s = 2; // No SNTP sync since boot yet
+            } else {
+                uint32_t ageSec = (uint32_t) ((millis64() - lastSync) / 1000);
+                s = ageSec > NTP_STALE_AFTER_SECONDS ? 2 : 1;
+            }
+            appendServiceEntry(out, "ntp", s, 0, server, NULL);
+        }
+    }
+
+    #if defined(AMS_CLOUD)
+    {
+        CloudConfig cc;
+        if(ctx.config->getCloudConfig(cc) && cc.enabled) {
+            uint8_t s;
+            int16_t err = ctx.cloud == NULL ? 0 : ctx.cloud->getLastError();
+            if(ctx.cloud == NULL || !ctx.cloud->isInitialized()) {
+                s = 2;
+            } else {
+                unsigned long since = millis() - ctx.cloud->getLastUpdate();
+                uint32_t maxAge = ((uint32_t) cc.interval) * 3000;
+                s = (ctx.cloud->getLastUpdate() > 0 && since > maxAge) ? 3 : 1;
+            }
+            appendServiceEntry(out, "cloud", s, err, cc.hostname, NULL);
+        }
+    }
+    #endif
+
+    #if defined(ZMART_CHARGE)
+    {
+        ZmartChargeConfig zc;
+        if(ctx.config->getZmartChargeConfig(zc) && zc.enabled) {
+            uint8_t s;
+            int16_t err = ctx.zcloud == NULL ? 0 : ctx.zcloud->getLastError();
+            if(ctx.zcloud == NULL || ctx.zcloud->getLastUpdate() == 0) {
+                s = 2;
+            } else {
+                s = ctx.zcloud->isLastFailed() ? 3 : 1;
+            }
+            appendServiceEntry(out, "zc", s, err, zc.baseUrl, NULL);
+        }
+    }
+    #endif
+
+    return out;
+}
 
 void AmsJsonGenerator::generateDayPlotJson(AmsDataStorage* ds, char* buf, size_t bufSize) {
 		uint16_t pos = snprintf_P(buf, bufSize, PSTR("{\"unit\":\"kwh\""));

src/AmsJsonGenerator.h

@@ -50,6 +50,38 @@ class BufferJsonSink : public JsonSink {
     bool overflowed_ = false;
 };
 
+class AmsData;
+class PriceService;
+class AmsMqttHandler;
+#if defined(AMS_CLOUD)
+class CloudConnector;
+#endif
+#if defined(ZMART_CHARGE)
+class ZmartChargeCloudConnector;
+#endif
+
+/**
+ * Everything the services array reports on. Both the web UI and the MQTT
+ * announcements render from this, so the two cannot report different state for
+ * the same device. Callers fill in whatever they have; NULL members are reported
+ * as not-yet-connected rather than skipped.
+ */
+struct ServiceStatusContext {
+    AmsConfiguration* config = NULL;
+    AmsData* meterState = NULL;
+    PriceService* ps = NULL;
+    AmsMqttHandler* mqttHandler = NULL;
+    bool mqttEnabled = false;
+    AmsMqttHandler* customMqttHandler = NULL;
+    AmsMqttHandler* energySpeedometer = NULL;
+    #if defined(AMS_CLOUD)
+    CloudConnector* cloud = NULL;
+    #endif
+    #if defined(ZMART_CHARGE)
+    ZmartChargeCloudConnector* zcloud = NULL;
+    #endif
+};
+
 class AmsJsonGenerator {
 public:
     static void generateDayPlotJson(AmsDataStorage* ds, char* buf, size_t bufSize);
@@ -58,4 +90,16 @@ class AmsJsonGenerator {
     static void generateConfigurationJson(AmsConfiguration* config, JsonSink& sink);
     // Convenience wrapper for callers that need the whole document in one buffer.
     static void generateConfigurationJson(AmsConfiguration* config, char* buf, size_t bufSize);
+
+    // Four-state health of a single service: 0 disabled, 1 ok, 2 connecting, 3 error.
+    static uint8_t hanState(AmsData* meterState);
+    static uint8_t mqttHandlerState(AmsMqttHandler* h);
+
+    // The contents of the services array, without the enclosing brackets, since
+    // both callers embed it in a larger document.
+    //
+    // Returns a String rather than streaming to a JsonSink: the array is small,
+    // both call sites need it as one contiguous value, and a fixed buffer would
+    // risk truncating to invalid JSON when hostnames are long.
+    static String generateServicesJson(const ServiceStatusContext& ctx);
 };