1.4.0: MQTT with Home Assistant, German SML, Belgian and Luxembourg P1, water and thermal subdevices

AGENTS.md

@@ -45,8 +45,16 @@ For a sink with its own storage model (like ClickHouse or QuestDB):
 3. Wire it in `cmd/meterlogger/sinks.go` (`buildSourceSinks`); the compiler forces every
    source to provide a constructor for it.
 
-Either way, add the sink to the sink table in `documentation/README.md` and config examples
-in `documentation/configuration.md`.
+For a non-database sink (like stdout or MQTT): add a `sinkInit` row in
+`buildSourceSinks` and register the config struct, defaults, and validation in
+`internal/config/` (including the sink name constant and the "at least one sink" check
+in `Validate`). The MQTT sink additionally keeps one shared broker client per process in
+`cmd/meterlogger/mqtt.go`, closed from `runtime.go`; follow that pattern for other
+connection-holding non-DB sinks.
+
+Either way, add the sink to the sink table in `documentation/README.md` and the root
+`README.md`, and config examples in `documentation/configuration.md` and
+`config.example.yaml`.
 
 ## Adding a new source
 

CHANGELOG.md

@@ -6,6 +6,51 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+## [1.4.0] - 2026-08-02
+
+### Fixed
+
+- QuestDB heat power and maximum power were stored three orders of magnitude
+  too small: the writer still divided by the milliwatt units of the M-Bus
+  library used before the gombus migration. Existing QuestDB heat history
+  shows a step change at this release.
+
+### Added
+
+- Belgian (Fluvius eMUCS) grid meter support: version line on `0-0:96.1.4`,
+  gas subdevices on `0-n:24.2.3` (volume not temperature corrected), decimal
+  phase currents, and peak demand (capaciteitstarief) fields stored in three
+  new grid columns: `avg_demand`, `max_demand_month`, `max_demand_month_at`
+  (grid schema migration v2, added automatically).
+- Water and thermal meter readings from the grid meter's P1 port
+  (`Grid.Water.Enabled`, `Grid.Thermal.Enabled`), alongside the existing gas
+  support. Water meters (device types 6 and 7, common on Belgian Fluvius
+  installs) store to `Grid.Water.Measurement` (default `water_meter`);
+  heat and cooling meters (device types 4, 10, 11, 12) store to
+  `Grid.Thermal.Measurement` (default `thermal_meter`). Readings are
+  deduplicated on the meter-supplied capture time, exactly like gas. Slave
+  e-meters (device type 2) are never stored from the master's telegram: read
+  them from their own P1 port.
+- Encrypted DLMS telegram support for Luxembourgish Smarty and Austrian
+  Sagemcom T210-D meters (EVN, Energienetze Steiermark) via
+  `Grid.DecryptionKey` and `Grid.AuthenticationKey`. Frames are AES-128-GCM
+  decrypted and fed through the normal telegram path. Telegrams with energy
+  totals only (`1.8.0`/`2.8.0`) and the `0-0:42.0.0` equipment id are
+  accepted. Wiener Netze raw DLMS push is not supported.
+- SML reader for German electricity meters (EMH eHZ and mMe4.0, ISKRA
+  MT681, EasyMeter Q3A/Q3B, eBZ DD3 SM variant, Holley DTZ541) over an IR
+  read head, selected with `Grid.Reader: sml` (default stays `dsmr`).
+  Accepts both the standard X-25 frame CRC and the Holley Kermit variant.
+  Works with factory-state meters that send only the energy total.
+- MQTT sink (`MQTT.Enabled`) that publishes every reading as flat JSON on
+  `<TopicPrefix>/<measurement>` and announces all sensors to Home Assistant
+  via retained MQTT discovery messages: one device per meter, correct
+  device/state classes and units, availability via a last-will status topic.
+  Grid, gas, and solar sensors slot straight into the Home Assistant Energy
+  dashboard; heat energy is published in kWh (converted from joules) for the
+  same reason. Water and thermal subdevice readings are announced too. See documentation/deployment.md, section "Home Assistant".
+
+
 ## [1.3.0] - 2026-08-01
 
 ### Added

README.md

@@ -25,7 +25,8 @@ for each meter type you want to read.
 | Source        | Protocol                  | Connection                  |
 |---------------|---------------------------|-----------------------------|
 | `heat`        | M-Bus (EN 13757) or optical (Kamstrup KMP, Multical 401/66C) | USB M-Bus adapter or IR read head |
-| `grid`        | DSMR P1 Dutch smart meter, plus gas meter readings via the P1 M-Bus channel when enabled | USB-to-P1 serial cable |
+| `grid`        | DSMR P1 smart meter (NL, BE Fluvius, LU Smarty, AT Sagemcom T210-D) or German SML meter via IR read head, plus gas, water and thermal meter readings via the P1 M-Bus channels when enabled | USB-to-P1 serial cable or IR read head |
+
 | `solar`       | Enphase Envoy HTTP API    | Local network               |
 | `ventilation` | DucoBox HTTP API          | Local network               |
 
@@ -39,11 +40,15 @@ for each meter type you want to read.
 | TimescaleDB | PostgreSQL extension | yes               |
 | ClickHouse  | column-store, OLAP   | yes               |
 | TDEngine    | time-series, IoT     | yes               |
+| MQTT        | message broker, HA   | n/a               |
 | Stdout      | debug, logs records  | n/a               |
 
 At least one sink must be enabled. All enabled sinks receive every write.
 The stdout sink logs data instead of persisting it; enable it with
 `Stdout.Enabled: true` for debugging, not for production.
+The MQTT sink publishes every reading to a broker and announces the sensors
+to Home Assistant via MQTT discovery; see
+[deployment.md](documentation/deployment.md#home-assistant).
 
 ## Quick start
 

cmd/meterlogger/mqtt.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+
+	"github.com/yottabytesolutions/meterlogger/internal/adapters/sink/mqtt"
+	"github.com/yottabytesolutions/meterlogger/internal/healthserver"
+)
+
+// mqttConnect is a test seam over mqtt.NewClient so wiring tests can build
+// the MQTT sink row without a real broker.
+//
+//nolint:gochecknoglobals // test seam, mirrors osExit
+var mqttConnect = mqtt.NewClient
+
+// sharedMQTT holds the single broker connection for the whole process. Unlike
+// QuestDB, one MQTT client safely serves every source; the client serializes
+// publishes internally.
+//
+//nolint:gochecknoglobals // process-wide shared connection, guarded by its mutex
+var sharedMQTT struct {
+	mu     sync.Mutex
+	client *mqtt.Client
+	err    error
+	inited bool
+}
+
+// sharedMQTTClient lazily connects the process-wide MQTT client on first use
+// and registers it with the health server. Subsequent callers get the same
+// client (or the same connection error).
+func sharedMQTTClient(
+	ctx context.Context, l *slog.Logger, healthSrv *healthserver.Server,
+) (*mqtt.Client, error) {
+	sharedMQTT.mu.Lock()
+	defer sharedMQTT.mu.Unlock()
+	if sharedMQTT.inited {
+		return sharedMQTT.client, sharedMQTT.err
+	}
+	sharedMQTT.inited = true
+
+	client, err := mqttConnect(ctx, mqtt.Config{
+		BrokerURL:              cfg.MQTT.BrokerURL,
+		Username:               cfg.MQTT.Username,
+		Password:               cfg.MQTT.Password,
+		ClientID:               mqttClientID(),
+		TopicPrefix:            cfg.MQTT.TopicPrefix,
+		HomeAssistantDiscovery: cfg.MQTT.HomeAssistantDiscovery,
+		DiscoveryPrefix:        cfg.MQTT.DiscoveryPrefix,
+		QoS:                    byte(cfg.MQTT.QoS), //nolint:gosec // G115: validated to 0 or 1
+		RetainState:            cfg.MQTT.RetainState,
+	}, l)
+	if err != nil {
+		sharedMQTT.err = err
+		return nil, err
+	}
+	if healthSrv != nil && client != nil {
+		healthSrv.Register(client)
+	}
+	sharedMQTT.client = client
+	return client, nil
+}
+
+// defaultMQTTClientID is the base MQTT client id when none is configured.
+const defaultMQTTClientID = "meterlogger"
+
+// mqttClientID resolves the configured client id, defaulting to "meterlogger"
+// suffixed with the --source filter so the one-container-per-source model
+// gets a unique id per process out of the box.
+func mqttClientID() string {
+	if cfg.MQTT.ClientID != "" {
+		return cfg.MQTT.ClientID
+	}
+	if sourceFilter != "" {
+		return defaultMQTTClientID + "-" + sourceFilter
+	}
+	return defaultMQTTClientID
+}
+
+// closeMQTT publishes the retained offline status and disconnects the shared
+// client, if one was ever created.
+func closeMQTT() {
+	sharedMQTT.mu.Lock()
+	defer sharedMQTT.mu.Unlock()
+	if sharedMQTT.client != nil {
+		if err := sharedMQTT.client.Close(); err != nil {
+			logger.Error("mqtt close error", slog.Any("error", err))
+		}
+	}
+	sharedMQTT.client, sharedMQTT.err, sharedMQTT.inited = nil, nil, false
+}

cmd/meterlogger/mqtt_test.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"testing"
+
+	"github.com/yottabytesolutions/meterlogger/internal/adapters/sink/mqtt"
+	"github.com/yottabytesolutions/meterlogger/internal/config"
+)
+
+// resetSharedMQTT clears the process-wide MQTT state around a test and stubs
+// the connect seam with the given function.
+func resetSharedMQTT(t *testing.T, connect func(context.Context, mqtt.Config, *slog.Logger) (*mqtt.Client, error)) {
+	t.Helper()
+	origConnect := mqttConnect
+	mqttConnect = connect
+	sharedMQTT.client, sharedMQTT.err, sharedMQTT.inited = nil, nil, false
+	t.Cleanup(func() {
+		mqttConnect = origConnect
+		sharedMQTT.client, sharedMQTT.err, sharedMQTT.inited = nil, nil, false
+	})
+}
+
+func TestMQTTClientID(t *testing.T) {
+	origCfg, origFilter := cfg, sourceFilter
+	defer func() { cfg, sourceFilter = origCfg, origFilter }()
+
+	tests := []struct {
+		name     string
+		clientID string
+		filter   string
+		want     string
+	}{
+		{"default", "", "", defaultMQTTClientID},
+		{"default with source filter", "", config.SourceGrid, defaultMQTTClientID + "-" + config.SourceGrid},
+		{"explicit id wins over filter", "custom", config.SourceGrid, "custom"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg = config.Config{MQTT: config.MQTTConfig{ClientID: tt.clientID}}
+			sourceFilter = tt.filter
+			if got := mqttClientID(); got != tt.want {
+				t.Errorf("mqttClientID() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestBuildSourceSinks_MQTTOnly(t *testing.T) {
+	origCfg := cfg
+	cfg = config.Config{MQTT: config.MQTTConfig{Enabled: true, BrokerURL: "tcp://broker:1883"}}
+	defer func() { cfg = origCfg }()
+
+	connects := 0
+	resetSharedMQTT(t, func(context.Context, mqtt.Config, *slog.Logger) (*mqtt.Client, error) {
+		connects++
+		return &mqtt.Client{}, nil
+	})
+
+	ctx := context.Background()
+	l := testLogger()
+	var dbs dbConnections
+
+	if got := len(buildHeatSinks(ctx, l, nil, dbs)); got != 1 {
+		t.Errorf("heat sinks = %d, want 1", got)
+	}
+	if got := len(buildGridSinks(ctx, l, nil, dbs)); got != 1 {
+		t.Errorf("grid sinks = %d, want 1", got)
+	}
+	if got := len(buildGasSinks(ctx, l, nil, dbs)); got != 1 {
+		t.Errorf("gas sinks = %d, want 1", got)
+	}
+	if got := len(buildSolarSinks(ctx, l, nil, dbs)); got != 1 {
+		t.Errorf("solar sinks = %d, want 1", got)
+	}
+	if got := len(buildVentilationSinks(ctx, l, nil, dbs)); got != 1 {
+		t.Errorf("ventilation sinks = %d, want 1", got)
+	}
+	if connects != 1 {
+		t.Errorf("mqtt connects = %d, want 1 shared connection", connects)
+	}
+}
+
+func TestSharedMQTTClient_CachesConnectionError(t *testing.T) {
+	origCfg := cfg
+	cfg = config.Config{MQTT: config.MQTTConfig{Enabled: true, BrokerURL: "tcp://broker:1883"}}
+	defer func() { cfg = origCfg }()
+
+	connects := 0
+	resetSharedMQTT(t, func(context.Context, mqtt.Config, *slog.Logger) (*mqtt.Client, error) {
+		connects++
+		return nil, errors.New("connection refused")
+	})
+
+	ctx := context.Background()
+	if _, err := sharedMQTTClient(ctx, testLogger(), nil); err == nil {
+		t.Fatal("first call should return the connection error")
+	}
+	if _, err := sharedMQTTClient(ctx, testLogger(), nil); err == nil {
+		t.Fatal("second call should return the cached connection error")
+	}
+	if connects != 1 {
+		t.Errorf("connect attempts = %d, want 1", connects)
+	}
+}
+
+func TestCloseMQTT_NoClientIsNoop(t *testing.T) {
+	resetSharedMQTT(t, mqttConnect)
+	closeMQTT() // must not panic
+	if sharedMQTT.inited {
+		t.Error("closeMQTT should reset the shared state")
+	}
+}

cmd/meterlogger/probe.go

@@ -15,7 +15,6 @@ import (
 
 	"github.com/yottabytesolutions/meterlogger/internal/adapters/source/ducobox"
 	"github.com/yottabytesolutions/meterlogger/internal/adapters/source/enphase"
-	"github.com/yottabytesolutions/meterlogger/internal/adapters/source/gridmeter"
 	"github.com/yottabytesolutions/meterlogger/internal/config"
 	"github.com/yottabytesolutions/meterlogger/internal/domain"
 )
@@ -137,7 +136,11 @@ func probeGrid(ctx context.Context, l *slog.Logger) (json.RawMessage, error) {
 	if cfg.Grid.SerialInterface == "" {
 		return nil, errors.New("grid source not configured: Grid.SerialInterface is empty")
 	}
-	telegram, err := readOneGridTelegram(ctx, gridmeter.NewGridReader(cfg.Grid.SerialInterface, l))
+	reader, err := newGridReader(l)
+	if err != nil {
+		return nil, err
+	}
+	telegram, err := readOneGridTelegram(ctx, reader)
 	if err != nil {
 		return nil, err
 	}

cmd/meterlogger/runtime.go

@@ -94,8 +94,10 @@ func (rt *app) startSources(ctx context.Context) {
 }
 
 // shutdown releases everything newRuntime assembled, in reverse order: sink
-// connections first, then profiling, then tracing.
+// connections first, then profiling, then tracing. The MQTT client goes first
+// so its retained offline status reaches the broker before teardown.
 func (rt *app) shutdown() {
+	closeMQTT()
 	closeAll(rt.dbs.closers())
 	if err := rt.stopProfiling(); err != nil {
 		logger.Error("failed to stop profiling", slog.Any("error", err))

cmd/meterlogger/sinks.go

@@ -7,6 +7,7 @@ import (
 	"os"
 
 	"github.com/yottabytesolutions/meterlogger/internal/adapters/sink/clickhouse"
+	"github.com/yottabytesolutions/meterlogger/internal/adapters/sink/mqtt"
 	"github.com/yottabytesolutions/meterlogger/internal/adapters/sink/qdb"
 	"github.com/yottabytesolutions/meterlogger/internal/adapters/sink/sqlsink"
 	"github.com/yottabytesolutions/meterlogger/internal/adapters/sink/stdout"
@@ -33,6 +34,7 @@ func buildSourceSinks[R any](
 	dbs dbConnections,
 	measurement string,
 	newQuestDBWriter func(client *qdb.DBClient, measurement string, l *slog.Logger) R,
+	newMQTTWriter func(client *mqtt.Client, measurement string, l *slog.Logger) R,
 	newSQLStore func(ctx context.Context, db *sqlsink.DB, measurement string, l *slog.Logger) (R, error),
 	newClickHouseStore func(ctx context.Context, db *clickhouse.DB, measurement string, l *slog.Logger) (R, error),
 ) []R {
@@ -55,6 +57,14 @@ func buildSourceSinks[R any](
 			}
 			return sink, nil
 		}},
+		{config.SinkMQTT, cfg.MQTT.Enabled, func() (R, error) {
+			client, err := sharedMQTTClient(ctx, l, healthSrv)
+			if err != nil {
+				var zero R
+				return zero, err
+			}
+			return newMQTTWriter(client, measurement, l), nil
+		}},
 	}
 	for _, db := range dbs.sql() {
 		inits = append(inits, sinkInit[R]{db.Name(), true, func() (R, error) {