Collect 15min energy metrics

core/metrics/accumulator.go

@@ -0,0 +1,120 @@
+package metrics
+
+import (
+	"bytes"
+	"fmt"
+	"time"
+
+	"github.com/benbjohnson/clock"
+	"github.com/samber/lo"
+)
+
+type Accumulator struct {
+	clock              clock.Clock
+	updated            time.Time
+	posMeter, negMeter *float64 // kWh
+	Pos                float64  `json:"pos"` // kWh
+	Neg                float64  `json:"neg"` // kWh
+}
+
+func WithClock(clock clock.Clock) func(*Accumulator) {
+	return func(m *Accumulator) {
+		m.clock = clock
+	}
+}
+
+func NewAccumulator(opt ...func(*Accumulator)) *Accumulator {
+	m := &Accumulator{clock: clock.New()}
+	for _, o := range opt {
+		o(m)
+	}
+	return m
+}
+
+func (m *Accumulator) Updated() time.Time {
+	return m.updated
+}
+
+func (m *Accumulator) String() string {
+	b := new(bytes.Buffer)
+	fmt.Fprintf(b, "Accumulated: %.3fkWh pos, %.3fkWh neg, updated: %v", m.Pos, m.Neg, m.updated.Truncate(time.Second))
+	if m.posMeter != nil || m.negMeter != nil {
+		fmt.Fprintf(b, " meter: ")
+		if m.posMeter != nil {
+			fmt.Fprintf(b, " %.3fkWh pos", *m.posMeter)
+		}
+		if m.negMeter != nil {
+			fmt.Fprintf(b, " %.3fkWh pos", *m.negMeter)
+		}
+	}
+	return b.String()
+}
+
+// PosEnergy returns the accumulated energy in kWh
+func (m *Accumulator) PosEnergy() float64 {
+	return m.Pos
+}
+
+// NegEnergy returns the accumulated energy in kWh
+func (m *Accumulator) NegEnergy() float64 {
+	return m.Neg
+}
+
+// AddPosMeterTotal adds the difference to the last total meter value in kWh
+func (m *Accumulator) AddPosMeterTotal(v float64) {
+	defer func() {
+		m.updated = m.clock.Now()
+		m.posMeter = lo.ToPtr(v)
+	}()
+
+	if m.posMeter == nil {
+		return
+	}
+
+	m.Pos += v - *m.posMeter
+}
+
+// AddNegMeterTotal adds the difference to the last total meter value in kWh
+func (m *Accumulator) AddNegMeterTotal(v float64) {
+	defer func() {
+		m.updated = m.clock.Now()
+		m.negMeter = lo.ToPtr(v)
+	}()
+
+	if m.negMeter == nil {
+		return
+	}
+
+	m.Neg += v - *m.negMeter
+}
+
+// AddPosEnergy adds the given energy in kWh to the positive meter
+func (m *Accumulator) AddPosEnergy(v float64) {
+	defer func() { m.updated = m.clock.Now() }()
+
+	if m.updated.IsZero() {
+		return
+	}
+
+	m.Pos += v
+}
+
+// AddNegEnergy adds the given energy in kWh to the negative meter
+func (m *Accumulator) AddNegEnergy(v float64) {
+	defer func() { m.updated = m.clock.Now() }()
+
+	if m.updated.IsZero() {
+		return
+	}
+
+	m.Neg += v
+}
+
+// AddPower adds the given power in W, calculating the energy based on the time since the last update
+func (m *Accumulator) AddPower(v float64) {
+	if v >= 0 {
+		m.AddPosEnergy(v * m.clock.Since(m.updated).Hours() / 1e3)
+	} else {
+		m.AddNegEnergy(-v * m.clock.Since(m.updated).Hours() / 1e3)
+	}
+}

core/meterenergy_test.go → core/metrics/accumulator_test.go

@@ -1,4 +1,4 @@
-package core
+package metrics
 
 import (
 	"testing"
@@ -13,31 +13,31 @@ func TestMeterEnergyMeterTotal(t *testing.T) {
 	clock := clock.NewMock()
 	clock.Set(now.BeginningOfDay())
 
-	me := &meterEnergy{clock: clock}
+	me := &Accumulator{clock: clock}
 
-	me.AddMeterTotal(10)
-	assert.Equal(t, 0.0, me.AccumulatedEnergy())
-	me.AddMeterTotal(11)
-	assert.Equal(t, 1.0, me.AccumulatedEnergy())
-	me.AddMeterTotal(11)
-	assert.Equal(t, 1.0, me.AccumulatedEnergy())
+	me.AddPosMeterTotal(10)
+	assert.Equal(t, 0.0, me.PosEnergy())
+	me.AddPosMeterTotal(11)
+	assert.Equal(t, 1.0, me.PosEnergy())
+	me.AddPosMeterTotal(11)
+	assert.Equal(t, 1.0, me.PosEnergy())
 }
 
 func TestMeterEnergyAddPower(t *testing.T) {
 	clock := clock.NewMock()
 	clock.Set(now.BeginningOfDay())
 
-	me := &meterEnergy{clock: clock}
+	me := &Accumulator{clock: clock}
 
 	clock.Add(60 * time.Minute)
 	me.AddPower(1e3)
-	assert.Equal(t, 0.0, me.AccumulatedEnergy())
+	assert.Equal(t, 0.0, me.PosEnergy())
 
 	clock.Add(60 * time.Minute)
 	me.AddPower(1e3)
-	assert.Equal(t, 1.0, me.AccumulatedEnergy())
+	assert.Equal(t, 1.0, me.PosEnergy())
 
 	clock.Add(30 * time.Minute)
 	me.AddPower(1e3)
-	assert.Equal(t, 1.5, me.AccumulatedEnergy())
+	assert.Equal(t, 1.5, me.PosEnergy())
 }

core/metrics/collector.go

@@ -0,0 +1,113 @@
+package metrics
+
+import (
+	"time"
+
+	"github.com/evcc-io/evcc/server/db"
+)
+
+const (
+	SlotDuration = 15 * time.Minute
+
+	// groups
+	Virtual = "virtual"
+	Grid    = "grid"
+	PV      = "pv"
+
+	// meters
+	Home = "home" // virtual home meter
+)
+
+type Collector struct {
+	entity  entity
+	accu    *Accumulator
+	started time.Time
+}
+
+func NewCollector(group, name string, opt ...func(*Accumulator)) (*Collector, error) {
+	entity, err := createEntity(group, name)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Collector{
+		entity: entity,
+		accu:   NewAccumulator(opt...),
+	}, nil
+}
+
+func createEntity(group, name string) (entity, error) {
+	entity := entity{
+		Group: group,
+		Name:  name,
+	}
+
+	if err := db.Instance.Where(&entity).FirstOrCreate(&entity).Error; err != nil {
+		return entity, err
+	}
+
+	return entity, nil
+}
+
+func (c *Collector) process(fun func()) error {
+	now := c.accu.clock.Now()
+
+	if c.accu.updated.IsZero() {
+		c.started = now
+	}
+
+	fun()
+
+	if slotStart := now.Truncate(SlotDuration); slotStart.After(c.started) {
+		// full slot completed
+		if slotStart.Sub(c.started) == SlotDuration {
+			if err := c.persist(); err != nil {
+				return err
+			}
+		}
+
+		c.started = slotStart
+		c.accu.Pos = 0
+		c.accu.Neg = 0
+	}
+
+	return nil
+}
+
+func (c *Collector) persist() error {
+	return persist(c.entity, c.started, c.accu.PosEnergy(), c.accu.NegEnergy())
+}
+
+func (c *Collector) Profile(from time.Time) (*[96]float64, error) {
+	return profile(c.entity, from)
+}
+
+func (c *Collector) AddPosEnergy(v float64) error {
+	return c.process(func() {
+		c.accu.AddPosEnergy(v)
+	})
+}
+
+func (c *Collector) AddNegEnergy(v float64) error {
+	return c.process(func() {
+		c.accu.AddNegEnergy(v)
+	})
+}
+
+func (c *Collector) AddPosMeterTotal(v float64) error {
+	return c.process(func() {
+		c.accu.AddPosMeterTotal(v)
+	})
+}
+
+func (c *Collector) AddNegMeterTotal(v float64) error {
+	return c.process(func() {
+		c.accu.AddNegMeterTotal(v)
+	})
+}
+
+func (c *Collector) AddPower(v float64) error {
+	return c.process(func() {
+		c.accu.AddPower(v)
+	})
+}

core/metrics/collector_test.go

@@ -0,0 +1,37 @@
+package metrics
+
+import (
+	"testing"
+	"time"
+
+	"github.com/benbjohnson/clock"
+	"github.com/evcc-io/evcc/server/db"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCollectorAddPower(t *testing.T) {
+	clock := clock.NewMock()
+
+	require.NoError(t, db.NewInstance("sqlite", ":memory:"))
+	Init()
+
+	col, err := NewCollector("foo", "foo", WithClock(clock))
+	require.NoError(t, err)
+	require.True(t, col.accu.updated.IsZero())
+
+	clock.Add(5 * time.Minute)
+	require.NoError(t, col.AddPower(1e3))
+	require.False(t, col.accu.updated.IsZero())
+
+	clock.Add(5 * time.Minute)
+	require.NoError(t, col.AddPower(1e3))
+	require.Equal(t, 1e3*5/60/1e3, col.accu.PosEnergy()) // kWh
+
+	clock.Add(5 * time.Minute)
+	require.NoError(t, col.AddPower(1e3))
+	require.Equal(t, 0.0, col.accu.PosEnergy()) // accumulator reset after 15 minutes
+
+	clock.Add(15 * time.Minute)
+	require.NoError(t, col.AddPower(1e3))
+	require.Equal(t, 0.0, col.accu.PosEnergy()) // accumulator reset after 15 minutes
+}