Metrics: fall back to power integration when a meter loses its energy register (#33103)

core/loadpoint.go

@@ -273,8 +273,14 @@ func NewLoadpointFromConfig(log *util.Logger, settings settings.Settings, collec
 
 	lp.configureChargerType(lp.charger)
 	// add collector
-	if lp.chargeMeter != nil {
+	if lp.chargeMeter != nil && collector != nil {
 		lp.chargeEnergy = collector
+
+		// drop stale readings when the meter no longer reports totals
+		energy, returnEnergy := api.HasCap[api.MeterEnergy](lp.chargeMeter), api.HasCap[api.MeterReturnEnergy](lp.chargeMeter)
+		if err := collector.SetCapabilities(energy, returnEnergy); err != nil {
+			return lp, err
+		}
 	}
 
 	// set title after collector is wired to refresh the metrics entity

core/metrics/collector.go

@@ -202,6 +202,35 @@ func (c *Collector) SetEnergy(energy float64) error {
 	return nil
 }
 
+// SetCapabilities drops the persisted reading for a direction the device no longer
+// reports, so its energy falls back to power integration instead of freezing.
+func (c *Collector) SetCapabilities(energy, returnEnergy bool) error {
+	cols := make(map[string]any, 2)
+
+	// keyed on the entity, since an incomplete state is left unrestored and would
+	// otherwise resurface once the other direction is checkpointed again
+	if !energy && c.entity.EnergyMeter != nil {
+		c.accu.energyMeter = nil
+		c.entity.EnergyMeter = nil
+		cols["energy_meter"] = nil
+	}
+	if !returnEnergy && c.entity.ReturnEnergyMeter != nil {
+		c.accu.returnEnergyMeter = nil
+		c.entity.ReturnEnergyMeter = nil
+		cols["return_energy_meter"] = nil
+	}
+
+	if len(cols) == 0 {
+		return nil
+	}
+
+	// a surviving reading still covers the downtime for its own direction, so
+	// keep the restore rather than discarding that delta with the cleared one
+	c.restored = c.accu.energyMeter != nil || c.accu.returnEnergyMeter != nil
+
+	return db.Instance.Model(&c.entity).UpdateColumns(cols).Error
+}
+
 func (c *Collector) SetEnergyMeterTotal(v float64) error {
 	return c.process(func() {
 		c.accu.SetEnergyMeterTotal(v)

core/metrics/collector_test.go

@@ -600,3 +600,112 @@ func TestCollectorSetEnergy(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, 1.5, v)
 }
+
+// TestCollectorFallsBackToPowerAfterCapabilityLoss verifies that a meter which
+// lost its energy register stops using the stale checkpoint and integrates power
+// instead (https://github.com/evcc-io/evcc/issues/33091).
+func TestCollectorFallsBackToPowerAfterCapabilityLoss(t *testing.T) {
+	clk := clock.NewMock() // 1970-01-01 00:00:00 UTC, on a slot boundary
+
+	require.NoError(t, db.NewInstance("sqlite", ":memory:"))
+	require.NoError(t, SetupSchema())
+
+	col, err := NewCollector(PV, "capability", "", WithClock(clk))
+	require.NoError(t, err)
+
+	// meter reports totals, checkpoint is persisted at the slot boundary
+	require.NoError(t, col.AddEnergy(new(54716.0), nil, 1e3))
+	clk.Add(15 * time.Minute) // 00:15
+	require.NoError(t, col.AddEnergy(new(54716.0), nil, 1e3))
+
+	var e entity
+	require.NoError(t, db.Instance.First(&e, col.entity.Id).Error)
+	require.Equal(t, 54716.0, *e.EnergyMeter)
+
+	// restart without the energy register: stale reading would freeze energy
+	col2, err := NewCollector(PV, "capability", "", WithClock(clk))
+	require.NoError(t, err)
+	require.True(t, col2.restored)
+
+	require.NoError(t, col2.SetCapabilities(false, false))
+	require.Nil(t, col2.accu.energyMeter)
+	require.False(t, col2.restored, "no readings left to seed a restore")
+
+	require.NoError(t, db.Instance.First(&e, col2.entity.Id).Error)
+	require.Nil(t, e.EnergyMeter, "stale checkpoint must be cleared")
+
+	// power is integrated again
+	require.NoError(t, col2.AddEnergy(nil, nil, 1e3))
+	clk.Add(5 * time.Minute)
+	require.NoError(t, col2.AddEnergy(nil, nil, 1e3))
+	require.InDelta(t, 1e3*5/60/1e3, col2.accu.Energy, 1e-10)
+}
+
+// TestCollectorClearsUnrestoredCheckpoint verifies that a bidirectional group
+// also drops a checkpoint that was too incomplete to be restored, so it cannot
+// resurface once the other direction is checkpointed again.
+func TestCollectorClearsUnrestoredCheckpoint(t *testing.T) {
+	clk := clock.NewMock() // 1970-01-01 00:00:00 UTC, on a slot boundary
+
+	require.NoError(t, db.NewInstance("sqlite", ":memory:"))
+	require.NoError(t, SetupSchema())
+
+	col, err := NewCollector(Grid, "incomplete", "", WithClock(clk))
+	require.NoError(t, err)
+
+	// only the energy direction is metered, so the state stays incomplete
+	require.NoError(t, col.AddEnergy(new(100.0), nil, 0))
+	clk.Add(15 * time.Minute) // 00:15
+	require.NoError(t, col.AddEnergy(new(100.5), nil, 0))
+
+	col2, err := NewCollector(Grid, "incomplete", "", WithClock(clk))
+	require.NoError(t, err)
+	require.False(t, col2.restored, "incomplete state must not restore")
+	require.Nil(t, col2.accu.energyMeter)
+
+	require.NoError(t, col2.SetCapabilities(false, false))
+
+	var e entity
+	require.NoError(t, db.Instance.First(&e, col2.entity.Id).Error)
+	require.Nil(t, e.EnergyMeter, "unrestored checkpoint must be cleared too")
+}
+
+// TestCollectorKeepsRestoreForSurvivingDirection verifies that clearing one
+// direction of a bidirectional group does not discard the downtime delta the
+// other direction still covers.
+func TestCollectorKeepsRestoreForSurvivingDirection(t *testing.T) {
+	clk := clock.NewMock() // 1970-01-01 00:00:00 UTC, on a slot boundary
+
+	require.NoError(t, db.NewInstance("sqlite", ":memory:"))
+	require.NoError(t, SetupSchema())
+
+	col, err := NewCollector(Grid, "surviving", "", WithClock(clk))
+	require.NoError(t, err)
+
+	// checkpoint both directions
+	require.NoError(t, col.AddEnergy(new(100.0), new(200.0), 0))
+	clk.Add(15 * time.Minute) // 00:15
+	require.NoError(t, col.AddEnergy(new(100.5), new(200.2), 0))
+	clk.Add(15 * time.Minute) // 00:30
+	require.NoError(t, col.AddEnergy(new(101.0), new(200.4), 0))
+
+	// restart after 1h downtime, joining slot 01:30 mid-way, without export
+	clk.Add(65 * time.Minute) // 01:35
+	col2, err := NewCollector(Grid, "surviving", "", WithClock(clk))
+	require.NoError(t, err)
+	require.NoError(t, col2.SetCapabilities(true, false))
+	require.True(t, col2.restored, "the surviving import reading still seeds a restore")
+	require.Nil(t, col2.accu.returnEnergyMeter)
+
+	// the import delta across the downtime is kept
+	require.NoError(t, col2.AddEnergy(new(111.0), nil, 0))
+	require.InDelta(t, 10.0, col2.accu.Energy, 1e-10)
+
+	clk.Add(10 * time.Minute) // 01:45
+	require.NoError(t, col2.AddEnergy(new(111.0), nil, 0))
+
+	var m meter
+	require.NoError(t, db.Instance.Where("meter = ? AND ts = ?", col2.entity.Id, 90*60).First(&m).Error)
+	require.InDelta(t, 10.0, m.Energy, 1e-10)
+	require.True(t, m.Recovered, "catchup slot must be flagged recovered")
+}

core/site.go

@@ -186,6 +186,24 @@ func activeMeters(refs []string) ([]config.Device[api.Meter], error) {
 	return res, nil
 }
 
+// newMeterCollector creates a meter collector and reconciles the persisted meter
+// readings with the device's capabilities, so a device that lost its energy
+// registers falls back to power integration instead of freezing.
+func newMeterCollector(group, ref, title string, meter api.Meter) (*metrics.Collector, error) {
+	energy, returnEnergy := api.HasCap[api.MeterEnergy](meter), api.HasCap[api.MeterReturnEnergy](meter)
+	if group == metrics.Battery {
+		// batteries map discharge to energy, see updateBatteryMeters
+		energy, returnEnergy = returnEnergy, energy
+	}
+
+	c, err := metrics.NewCollector(group, ref, title)
+	if err != nil {
+		return nil, err
+	}
+
+	return c, c.SetCapabilities(energy, returnEnergy)
+}
+
 func (site *Site) Boot(log *util.Logger, loadpoints []*Loadpoint, tariffs *tariff.Tariffs) error {
 	site.loadpoints = loadpoints
 	site.tariffs = tariffs
@@ -252,7 +270,7 @@ func (site *Site) Boot(log *util.Logger, loadpoints []*Loadpoint, tariffs *tarif
 		} else {
 			site.gridMeter = dev
 
-			me, err := metrics.NewCollector(metrics.Grid, site.Meters.GridMeterRef, metrics.Grid)
+			me, err := newMeterCollector(metrics.Grid, site.Meters.GridMeterRef, metrics.Grid, dev.Instance())
 			if err != nil {
 				return err
 			}
@@ -272,7 +290,7 @@ func (site *Site) Boot(log *util.Logger, loadpoints []*Loadpoint, tariffs *tarif
 		site.pvMeters = append(site.pvMeters, dev)
 
 		// energy collector (for history persistence and forecast scaling)
-		me, err := metrics.NewCollector(metrics.PV, ref, deviceTitleOrName(dev))
+		me, err := newMeterCollector(metrics.PV, ref, deviceTitleOrName(dev), dev.Instance())
 		if err != nil {
 			return err
 		}
@@ -301,7 +319,7 @@ func (site *Site) Boot(log *util.Logger, loadpoints []*Loadpoint, tariffs *tarif
 	site.batteryMeters = mm
 	for _, dev := range mm {
 		ref := dev.Config().Name
-		me, err := metrics.NewCollector(metrics.Battery, ref, deviceTitleOrName(dev))
+		me, err := newMeterCollector(metrics.Battery, ref, deviceTitleOrName(dev), dev.Instance())
 		if err != nil {
 			return err
 		}
@@ -316,7 +334,7 @@ func (site *Site) Boot(log *util.Logger, loadpoints []*Loadpoint, tariffs *tarif
 	site.extMeters = mm
 	for _, dev := range mm {
 		ref := dev.Config().Name
-		me, err := metrics.NewCollector(metrics.Meter, ref, deviceTitleOrName(dev))
+		me, err := newMeterCollector(metrics.Meter, ref, deviceTitleOrName(dev), dev.Instance())
 		if err != nil {
 			return err
 		}
@@ -331,7 +349,7 @@ func (site *Site) Boot(log *util.Logger, loadpoints []*Loadpoint, tariffs *tarif
 	site.auxMeters = mm
 	for _, dev := range mm {
 		ref := dev.Config().Name
-		me, err := metrics.NewCollector(metrics.Consumer, ref, deviceTitleOrName(dev))
+		me, err := newMeterCollector(metrics.Consumer, ref, deviceTitleOrName(dev), dev.Instance())
 		if err != nil {
 			return err
 		}
@@ -346,7 +364,7 @@ func (site *Site) Boot(log *util.Logger, loadpoints []*Loadpoint, tariffs *tarif
 	site.consumerMeters = mm
 	for _, dev := range mm {
 		ref := dev.Config().Name
-		me, err := metrics.NewCollector(metrics.Consumer, ref, deviceTitleOrName(dev))
+		me, err := newMeterCollector(metrics.Consumer, ref, deviceTitleOrName(dev), dev.Instance())
 		if err != nil {
 			return err
 		}