diff --git a/cmd/root.go b/cmd/root.go index c33787b98..f92f77f10 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -135,6 +135,8 @@ func checkVersion() { var teeIsChained bool // controles piping of first channel in teed chain +// tee splits a tee channel from an input channel and starts a goroutine +// for duplicateing channel values to replacement input and tee channel func tee(in chan core.Param) (chan core.Param, <-chan core.Param) { gen := make(chan core.Param) tee := make(chan core.Param) @@ -205,6 +207,14 @@ func run(cmd *cobra.Command, args []string) { var teeChan <-chan core.Param valueChan, teeChan = tee(valueChan) + // eliminate duplicate values + dedupe := server.NewDeduplicator(30*time.Minute, "socCharge") + teeChan = dedupe.Pipe(teeChan) + + // reduce number of values written to influx + limiter := server.NewLimiter(1 * time.Minute) + teeChan = limiter.Pipe(teeChan) + go influx.Run(teeChan) } @@ -214,6 +224,7 @@ func run(cmd *cobra.Command, args []string) { var teeChan <-chan core.Param valueChan, teeChan = tee(valueChan) + go socketHub.Run(teeChan, triggerChan) // start all loadpoints diff --git a/evcc.dist.yaml b/evcc.dist.yaml index 5c6b01940..55d03729d 100644 --- a/evcc.dist.yaml +++ b/evcc.dist.yaml @@ -1,8 +1,16 @@ uri: 0.0.0.0:7070 # uri for ui interval: 10s # control cycle interval +# mqtt message broker mqtt: - broker: localhost:1883 # mqtt broker address + broker: localhost:1883 + # user: + # password: + +# influx database (v1) +influx: + url: http://localhost:8086 + database: evcc # user: # password: diff --git a/server/limiter.go b/server/limiter.go new file mode 100644 index 000000000..c1ecafaed --- /dev/null +++ b/server/limiter.go @@ -0,0 +1,99 @@ +package server + +import ( + "time" + + "github.com/andig/evcc/core" +) + +// Piper is the interface that data flow plugins must implement +type Piper interface { + Pipe(in <-chan core.Param) <-chan core.Param +} + +type cacheItem struct { + updated time.Time + val interface{} +} + +// Deduplicator allows filtering of channel data by given criteria +type Deduplicator struct { + interval time.Duration + filter map[string]interface{} + cache map[string]cacheItem +} + +// NewDeduplicator creates Deduplicator +func NewDeduplicator(interval time.Duration, filter ...string) Piper { + l := &Deduplicator{ + interval: interval, + filter: make(map[string]interface{}), + cache: make(map[string]cacheItem), + } + + for _, f := range filter { + l.filter[f] = struct{}{} + } + + return l +} + +func (l *Deduplicator) pipe(in <-chan core.Param, out chan<- core.Param) { + for p := range in { + // use loadpoint + param.Key as lookup key to value cache + key := p.LoadPoint + "." + p.Key + item, cached := l.cache[key] + _, filtered := l.filter[p.Key] + + // forward if not cached + if !cached || !filtered || filtered && + (time.Since(item.updated) >= l.interval || p.Val != item.val) { + l.cache[key] = cacheItem{updated: time.Now(), val: p.Val} + out <- p + } + } +} + +// Pipe creates a new filtered output channel for given input channel +func (l *Deduplicator) Pipe(in <-chan core.Param) <-chan core.Param { + out := make(chan core.Param) + go l.pipe(in, out) + return out +} + +// Limiter allows filtering of channel data by given criteria +type Limiter struct { + interval time.Duration + cache map[string]cacheItem +} + +// NewLimiter creates limiter +func NewLimiter(interval time.Duration) Piper { + l := &Limiter{ + interval: interval, + cache: make(map[string]cacheItem), + } + + return l +} + +func (l *Limiter) pipe(in <-chan core.Param, out chan<- core.Param) { + for p := range in { + // use loadpoint + param.Key as lookup key to value cache + key := p.LoadPoint + "." + p.Key + item, cached := l.cache[key] + + // forward if not cached or expired + if !cached || time.Since(item.updated) >= l.interval { + l.cache[key] = cacheItem{updated: time.Now(), val: p.Val} + out <- p + } + } +} + +// Pipe creates a new filtered output channel for given input channel +func (l *Limiter) Pipe(in <-chan core.Param) <-chan core.Param { + out := make(chan core.Param) + go l.pipe(in, out) + return out +}