From e2ea8f930d45e128ea29d0cd55e2593ade91ca9c Mon Sep 17 00:00:00 2001 From: andig Date: Thu, 23 Jul 2026 18:11:16 +0200 Subject: [PATCH 1/2] Metrics: track loadpoint enabled state per slot Record the time-weighted fraction (0..1) of each 15min slot during which a loadpoint was enabled, stored in a new nullable meters.enabled column. Non-loadpoint entities never sample it, so the column stays null for them. Enabled is integrated over time like power (AddEnabled), sampled in the same accumulator interval as the energy update so it shares the slot's timing. Recovered downtime slots store null since enabled was not sampled during the gap. --- core/loadpoint.go | 2 +- core/metrics/accumulator.go | 35 +++++++++++++++--- core/metrics/collector.go | 61 +++++++++++++++++++++----------- core/metrics/collector_test.go | 36 ++++++++++++++++++- core/metrics/db.go | 4 ++- core/metrics/db_entities_test.go | 4 +-- core/metrics/db_history_test.go | 4 +-- core/metrics/db_test.go | 28 +++++++-------- core/metrics/stats_test.go | 14 ++++---- 9 files changed, 135 insertions(+), 53 deletions(-) diff --git a/core/loadpoint.go b/core/loadpoint.go index deaddf4cf17..8fc727d7b67 100644 --- a/core/loadpoint.go +++ b/core/loadpoint.go @@ -1848,7 +1848,7 @@ func (lp *Loadpoint) publishChargeProgress() { } if lp.chargeEnergy != nil { - lp.chargeEnergy.AddEnergy(importTotal, nil, lp.chargePower) + lp.chargeEnergy.AddEnergyEnabled(importTotal, nil, lp.chargePower, lp.enabled) if v := lp.GetSoc(); v > 0 { lp.chargeEnergy.SetSocTemp(v, lp.chargerHasFeature(api.Heating)) } diff --git a/core/metrics/accumulator.go b/core/metrics/accumulator.go index 943af91b20b..0257ca853e7 100644 --- a/core/metrics/accumulator.go +++ b/core/metrics/accumulator.go @@ -11,11 +11,13 @@ import ( type Accumulator struct { clock clock.Clock updated time.Time - energyMeter *float64 // kWh - returnEnergyMeter *float64 // kWh - Energy float64 `json:"energy"` // kWh - ReturnEnergy float64 `json:"returnEnergy"` // kWh - SocTemp *float64 `json:"socTemp,omitempty"` + energyMeter *float64 // kWh + returnEnergyMeter *float64 // kWh + Energy float64 `json:"energy"` // kWh + ReturnEnergy float64 `json:"returnEnergy"` // kWh + SocTemp *float64 `json:"socTemp,omitempty"` + enabledTime time.Duration // time enabled within the current slot + totalTime time.Duration // sampled time within the current slot } // AccumulatorState is the resumable meter-reading checkpoint of an Accumulator. @@ -44,6 +46,29 @@ func (s AccumulatorState) CompleteFor(group string) bool { return s.EnergyMeter != nil || s.ReturnEnergyMeter != nil } +// AddEnabled integrates the enabled state over the interval since the last +// update, weighting by elapsed time like AddPower. +func (m *Accumulator) AddEnabled(enabled bool) { + if m.updated.IsZero() { + return + } + dt := m.clock.Since(m.updated) + m.totalTime += dt + if enabled { + m.enabledTime += dt + } +} + +// EnabledFraction returns the 0..1 fraction of the slot the loadpoint was +// enabled, or nil when the state was never sampled (non-loadpoint entities). +func (m *Accumulator) EnabledFraction() *float64 { + if m.totalTime <= 0 { + return nil + } + f := m.enabledTime.Seconds() / m.totalTime.Seconds() + return &f +} + // setSocTemp keeps the first reading per slot. func (m *Accumulator) setSocTemp(value float64) { if m.SocTemp == nil { diff --git a/core/metrics/collector.go b/core/metrics/collector.go index ac55601bcee..ea456ba5f4b 100644 --- a/core/metrics/collector.go +++ b/core/metrics/collector.go @@ -139,11 +139,18 @@ func (c *Collector) process(fun func()) error { c.accu.Energy = 0 c.accu.ReturnEnergy = 0 c.accu.SocTemp = nil + c.accu.enabledTime = 0 + c.accu.totalTime = 0 return nil } func (c *Collector) persist(recovered bool) error { - if err := persist(c.entity, c.started, c.accu.Energy, c.accu.ReturnEnergy, c.accu.SocTemp, recovered); err != nil { + // recovered slots are downtime catchup with no meaningful enabled sampling + enabled := c.accu.EnabledFraction() + if recovered { + enabled = nil + } + if err := persist(c.entity, c.started, c.accu.Energy, c.accu.ReturnEnergy, c.accu.SocTemp, enabled, recovered); err != nil { return err } @@ -201,26 +208,40 @@ func (c *Collector) SetReturnEnergyMeterTotal(v float64) error { // so a transient failure is recovered via the next delta and not double-counted. func (c *Collector) AddEnergy(energyTotal, returnEnergyTotal *float64, power float64) error { return c.process(func() { - // a direction that ever reported a total is metered, so a nil read is a - // transient failure rather than a power-only meter - hasEnergyMeter := energyTotal != nil || c.accu.energyMeter != nil - hasReturnMeter := returnEnergyTotal != nil || c.accu.returnEnergyMeter != nil - - // integrate power for the unmetered direction first, since applying a - // meter total advances the accumulator clock - if power >= 0 { - if !hasEnergyMeter { - c.accu.AddPower(power) - } - } else if !hasReturnMeter { + c.addEnergy(energyTotal, returnEnergyTotal, power) + }) +} + +// AddEnergyEnabled behaves like AddEnergy and additionally integrates the +// loadpoint enabled state into the slot. Enabled is sampled before the meter +// totals advance the accumulator clock, so it shares this cycle's interval. +func (c *Collector) AddEnergyEnabled(energyTotal, returnEnergyTotal *float64, power float64, enabled bool) error { + return c.process(func() { + c.accu.AddEnabled(enabled) + c.addEnergy(energyTotal, returnEnergyTotal, power) + }) +} + +func (c *Collector) addEnergy(energyTotal, returnEnergyTotal *float64, power float64) { + // a direction that ever reported a total is metered, so a nil read is a + // transient failure rather than a power-only meter + hasEnergyMeter := energyTotal != nil || c.accu.energyMeter != nil + hasReturnMeter := returnEnergyTotal != nil || c.accu.returnEnergyMeter != nil + + // integrate power for the unmetered direction first, since applying a + // meter total advances the accumulator clock + if power >= 0 { + if !hasEnergyMeter { c.accu.AddPower(power) } + } else if !hasReturnMeter { + c.accu.AddPower(power) + } - if energyTotal != nil { - c.accu.SetEnergyMeterTotal(*energyTotal) - } - if returnEnergyTotal != nil { - c.accu.SetReturnEnergyMeterTotal(*returnEnergyTotal) - } - }) + if energyTotal != nil { + c.accu.SetEnergyMeterTotal(*energyTotal) + } + if returnEnergyTotal != nil { + c.accu.SetReturnEnergyMeterTotal(*returnEnergyTotal) + } } diff --git a/core/metrics/collector_test.go b/core/metrics/collector_test.go index 0f4565e6985..27e38e939b7 100644 --- a/core/metrics/collector_test.go +++ b/core/metrics/collector_test.go @@ -489,7 +489,7 @@ func TestCreateEntityReconcilesExtToConsumer(t *testing.T) { // ext meter with a persisted history slot ext, err := createEntity(Meter, "db:5", "Fridge") require.NoError(t, err) - require.NoError(t, persist(ext, time.Unix(15*60, 0), 0.3, 0, nil, false)) + require.NoError(t, persist(ext, time.Unix(15*60, 0), 0.3, 0, nil, nil, false)) // reconfigured as consumer: same row relabeled, history intact con, err := createEntity(Consumer, "db:5", "Fridge") @@ -569,3 +569,37 @@ func TestCollectorLastSlotEnergy(t *testing.T) { _, ok = col.LastSlotEnergy() require.False(t, ok) } + +// TestCollectorEnabledFraction verifies the loadpoint enabled state is stored +// as the time-weighted 0..1 fraction of the slot, and stays nil for entities +// that never sample it. +func TestCollectorEnabledFraction(t *testing.T) { + clk := clock.NewMock() // starts at Unix 0, a slot boundary + + require.NoError(t, db.NewInstance("sqlite", ":memory:")) + require.NoError(t, SetupSchema()) + + lp, err := NewCollector(Loadpoint, "db:1", "", WithClock(clk)) + require.NoError(t, err) + grid, err := NewCollector(Grid, "db:2", "", WithClock(clk)) + require.NoError(t, err) + + // slot 0: enabled only for the first of three 5min intervals -> 1/3 + require.NoError(t, lp.AddEnergyEnabled(nil, nil, 1e3, true)) // t=0, seeds clock + require.NoError(t, grid.AddEnergy(nil, nil, 1e3)) + for _, enabled := range []bool{true, false, false} { + clk.Add(5 * time.Minute) + require.NoError(t, lp.AddEnergyEnabled(nil, nil, 1e3, enabled)) + require.NoError(t, grid.AddEnergy(nil, nil, 1e3)) + } + + var lpSlot meter + require.NoError(t, db.Instance.Where("meter = ? AND ts = 0", lp.entity.Id).First(&lpSlot).Error) + require.NotNil(t, lpSlot.Enabled) + require.InDelta(t, 1.0/3.0, *lpSlot.Enabled, 1e-9) + + // grid never samples enabled -> column stays null + var gridSlot meter + require.NoError(t, db.Instance.Where("meter = ? AND ts = 0", grid.entity.Id).First(&gridSlot).Error) + require.Nil(t, gridSlot.Enabled) +} diff --git a/core/metrics/db.go b/core/metrics/db.go index 47998ec6eb8..c210160cb4c 100644 --- a/core/metrics/db.go +++ b/core/metrics/db.go @@ -18,6 +18,7 @@ type meter struct { Energy float64 `json:"energy" gorm:"column:energy"` ReturnEnergy float64 `json:"returnEnergy" gorm:"column:return_energy"` SocTemp *float64 `json:"socTemp,omitempty" gorm:"column:soc_temp"` // at start of slot + Enabled *float64 `json:"enabled,omitempty" gorm:"column:enabled"` // fraction 0..1 of slot the loadpoint was enabled (loadpoints only) Recovered bool `json:"recovered,omitempty" gorm:"column:recovered"` // downtime catchup slot, excluded from profile } @@ -137,7 +138,7 @@ func SetupSchema() error { var OnPersist func(slot time.Time) // persist stores a completed 15min slot -func persist(entity entity, ts time.Time, energy, returnEnergy float64, socTemp *float64, recovered bool) error { +func persist(entity entity, ts time.Time, energy, returnEnergy float64, socTemp, enabled *float64, recovered bool) error { slot := ts.Truncate(tariff.SlotDuration) if err := db.Instance.Create(&meter{ Meter: entity.Id, @@ -145,6 +146,7 @@ func persist(entity entity, ts time.Time, energy, returnEnergy float64, socTemp Energy: energy, ReturnEnergy: returnEnergy, SocTemp: socTemp, + Enabled: enabled, Recovered: recovered, }).Error; err != nil { return err diff --git a/core/metrics/db_entities_test.go b/core/metrics/db_entities_test.go index 673e7d2145f..5e6b72aa9d4 100644 --- a/core/metrics/db_entities_test.go +++ b/core/metrics/db_entities_test.go @@ -19,8 +19,8 @@ func TestListEntities(t *testing.T) { require.NoError(t, db.Instance.Create(&pv).Error) base := time.Date(2026, 4, 15, 16, 0, 0, 0, time.Now().Location()) - require.NoError(t, persist(grid, base, 1, 0, nil, false)) - require.NoError(t, persist(grid, base.Add(time.Hour), 2, 0, nil, false)) + require.NoError(t, persist(grid, base, 1, 0, nil, nil, false)) + require.NoError(t, persist(grid, base.Add(time.Hour), 2, 0, nil, nil, false)) entities, err := ListEntities() require.NoError(t, err) diff --git a/core/metrics/db_history_test.go b/core/metrics/db_history_test.go index d2daf11a992..1a655f37250 100644 --- a/core/metrics/db_history_test.go +++ b/core/metrics/db_history_test.go @@ -175,8 +175,8 @@ func TestQueryEnergySoc(t *testing.T) { require.NoError(t, db.Instance.Create(&e).Error) base := time.Date(2026, 4, 15, 16, 0, 0, 0, time.Now().Location()) - require.NoError(t, persist(e, base, 1, 0, new(80.0), false)) - require.NoError(t, persist(e, base.Add(15*time.Minute), 1, 0, new(70.0), false)) + require.NoError(t, persist(e, base, 1, 0, new(80.0), nil, false)) + require.NoError(t, persist(e, base.Add(15*time.Minute), 1, 0, new(70.0), nil, false)) from := base.Add(-time.Hour).UTC() to := base.Add(time.Hour).UTC() diff --git a/core/metrics/db_test.go b/core/metrics/db_test.go index 5c66f1bd36d..f1fdc0cc5c6 100644 --- a/core/metrics/db_test.go +++ b/core/metrics/db_test.go @@ -20,7 +20,7 @@ func TestSqliteTimestamp(t *testing.T) { entity := entity{Name: "foo"} require.NoError(t, db.Instance.FirstOrCreate(&entity).Error) - persist(entity, clock.Now(), 0, 0, nil, false) + persist(entity, clock.Now(), 0, 0, nil, nil, false) db, err := db.Instance.DB() require.NoError(t, err) @@ -55,8 +55,8 @@ func TestQueryEnergyUTCFilter(t *testing.T) { loc := time.Now().Location() base := time.Date(2026, 4, 15, 16, 0, 0, 0, loc) - require.NoError(t, persist(e, base, 0, 1, nil, false)) - require.NoError(t, persist(e, base.Add(time.Hour), 0, 2, nil, false)) + require.NoError(t, persist(e, base, 0, 1, nil, nil, false)) + require.NoError(t, persist(e, base.Add(time.Hour), 0, 2, nil, nil, false)) // query with UTC times spanning both slots from := base.Add(-time.Hour).UTC() @@ -83,10 +83,10 @@ func TestQueryEnergyGrouped(t *testing.T) { loc := time.Now().Location() base := time.Date(2026, 4, 15, 16, 0, 0, 0, loc) - require.NoError(t, persist(e1, base, 1, 0, nil, false)) - require.NoError(t, persist(e2, base, 2, 0, nil, false)) - require.NoError(t, persist(e1, base.Add(time.Hour), 3, 0, nil, false)) - require.NoError(t, persist(e2, base.Add(time.Hour), 4, 0, nil, false)) + require.NoError(t, persist(e1, base, 1, 0, nil, nil, false)) + require.NoError(t, persist(e2, base, 2, 0, nil, nil, false)) + require.NoError(t, persist(e1, base.Add(time.Hour), 3, 0, nil, nil, false)) + require.NoError(t, persist(e2, base.Add(time.Hour), 4, 0, nil, nil, false)) from := base.Add(-time.Hour).UTC() to := base.Add(3 * time.Hour).UTC() @@ -125,9 +125,9 @@ func TestQueryEnergyMultipleSeries(t *testing.T) { // 2 hourly slots per entity for i := range 2 { ts := base.Add(time.Duration(i) * time.Hour) - require.NoError(t, persist(eGrid, ts, float64(1+i), 0, nil, false)) - require.NoError(t, persist(ePv1, ts, 0, float64(10+i), nil, false)) - require.NoError(t, persist(ePv2, ts, 0, float64(20+i), nil, false)) + require.NoError(t, persist(eGrid, ts, float64(1+i), 0, nil, nil, false)) + require.NoError(t, persist(ePv1, ts, 0, float64(10+i), nil, nil, false)) + require.NoError(t, persist(ePv2, ts, 0, float64(20+i), nil, nil, false)) } from := base.Add(-time.Hour).UTC() @@ -187,9 +187,9 @@ func TestQueryEnergyFilter(t *testing.T) { base := time.Date(2026, 4, 15, 16, 0, 0, 0, loc) for i := range 2 { ts := base.Add(time.Duration(i) * time.Hour) - require.NoError(t, persist(eGrid, ts, float64(1+i), 0, nil, false)) - require.NoError(t, persist(ePv, ts, 0, float64(10+i), nil, false)) - require.NoError(t, persist(eBat, ts, float64(5+i), 0, nil, false)) + require.NoError(t, persist(eGrid, ts, float64(1+i), 0, nil, nil, false)) + require.NoError(t, persist(ePv, ts, 0, float64(10+i), nil, nil, false)) + require.NoError(t, persist(eBat, ts, float64(5+i), 0, nil, nil, false)) } from := base.Add(-time.Hour).UTC() @@ -237,7 +237,7 @@ func TestUpdateProfile(t *testing.T) { // day 1: 0 ... 95 // day 2: 96 ... 181 for i := range 4 * 2 * 24 { - persist(entity, clock.Now(), float64(i), float64(i), nil, false) + persist(entity, clock.Now(), float64(i), float64(i), nil, nil, false) clock.Add(15 * time.Minute) } diff --git a/core/metrics/stats_test.go b/core/metrics/stats_test.go index 85653ff7e9a..877a1a17fe6 100644 --- a/core/metrics/stats_test.go +++ b/core/metrics/stats_test.go @@ -25,13 +25,13 @@ func TestCollectorEnergyStats(t *testing.T) { require.NoError(t, err) e := col.entity - require.NoError(t, persist(e, slotStart.AddDate(0, 0, -8), 100, 0, nil, false)) // outside 7d - require.NoError(t, persist(e, slotStart.Add(-24*time.Hour-15*time.Minute), 5, 0, nil, false)) // 7d only - require.NoError(t, persist(e, slotStart.Add(-24*time.Hour), 7, 0, nil, false)) // 24h window start - require.NoError(t, persist(e, slotStart.Add(-12*time.Hour), 3, 0, nil, false)) // yesterday, within 24h - require.NoError(t, persist(e, midnight, 1, 0, nil, false)) // first slot today - require.NoError(t, persist(e, slotStart.Add(-15*time.Minute), 2, 0, nil, false)) // last completed slot - require.NoError(t, persist(e, slotStart, 4, 0, nil, false)) // current slot, excluded + require.NoError(t, persist(e, slotStart.AddDate(0, 0, -8), 100, 0, nil, nil, false)) // outside 7d + require.NoError(t, persist(e, slotStart.Add(-24*time.Hour-15*time.Minute), 5, 0, nil, nil, false)) // 7d only + require.NoError(t, persist(e, slotStart.Add(-24*time.Hour), 7, 0, nil, nil, false)) // 24h window start + require.NoError(t, persist(e, slotStart.Add(-12*time.Hour), 3, 0, nil, nil, false)) // yesterday, within 24h + require.NoError(t, persist(e, midnight, 1, 0, nil, nil, false)) // first slot today + require.NoError(t, persist(e, slotStart.Add(-15*time.Minute), 2, 0, nil, nil, false)) // last completed slot + require.NoError(t, persist(e, slotStart, 4, 0, nil, nil, false)) // current slot, excluded stats, err := col.EnergyStats() require.NoError(t, err) From 515eccbfc8e5f3bae21fe1ee45d862a12ad0253d Mon Sep 17 00:00:00 2001 From: andig Date: Thu, 23 Jul 2026 18:29:07 +0200 Subject: [PATCH 2/2] Metrics: clamp enabled fraction to [0,1] contract --- core/metrics/accumulator.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/core/metrics/accumulator.go b/core/metrics/accumulator.go index 0257ca853e7..bf2be847802 100644 --- a/core/metrics/accumulator.go +++ b/core/metrics/accumulator.go @@ -65,7 +65,8 @@ func (m *Accumulator) EnabledFraction() *float64 { if m.totalTime <= 0 { return nil } - f := m.enabledTime.Seconds() / m.totalTime.Seconds() + // clamp to the documented 0..1 contract, robust to any future timing changes + f := max(0, min(1, m.enabledTime.Seconds()/m.totalTime.Seconds())) return &f }