-
Notifications
You must be signed in to change notification settings - Fork 111
/
movingAverage.go
146 lines (138 loc) · 3.38 KB
/
movingAverage.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
package library
import (
"container/heap"
"errors"
"github.com/nytlabs/gojee"
"github.com/nytlabs/streamtools/st/blocks" // blocks
"github.com/nytlabs/streamtools/st/util"
"time"
)
// specify those channels we're going to use to communicate with streamtools
type MovingAverage struct {
blocks.Block
queryrule chan chan interface{}
queryavg chan chan interface{}
inrule chan interface{}
inpoll chan interface{}
in chan interface{}
out chan interface{}
quit chan interface{}
}
// we need to build a simple factory so that streamtools can make new blocks of this kind
func NewMovingAverage() blocks.BlockInterface {
return &MovingAverage{}
}
// Setup is called once before running the block. We build up the channels and specify what kind of block this is.
func (b *MovingAverage) Setup() {
b.Kind = "MovingAverage"
b.Desc = "performs a moving average of the values specified by the Path over the duration of the Window"
b.in = b.InRoute("in")
b.inrule = b.InRoute("rule")
b.queryrule = b.QueryRoute("rule")
b.queryavg = b.QueryRoute("average")
b.inpoll = b.InRoute("poll")
b.quit = b.Quit()
b.out = b.Broadcast()
}
func pqAverage(pq *PriorityQueue) float64 {
var sum float64
sum = 0
for _, pqmsg := range *pq {
v := pqmsg.val
val, ok := v.(float64)
if !ok {
continue
}
sum += val
}
N := float64(len(*pq))
return sum / N
}
// Run is the block's main loop. Here we listen on the different channels we set up.
func (b *MovingAverage) Run() {
var tree *jee.TokenTree
var path, windowString string
var err error
window := time.Duration(0)
waitTimer := time.NewTimer(100 * time.Millisecond)
pq := &PriorityQueue{}
heap.Init(pq)
for {
select {
case ruleI := <-b.inrule:
// set a parameter of the block
path, err = util.ParseString(ruleI, "Path")
if err != nil {
b.Error(err)
}
tree, err = util.BuildTokenTree(path)
if err != nil {
b.Error(err)
break
}
windowString, err = util.ParseString(ruleI, "Window")
if err != nil {
b.Error(err)
}
window, err = time.ParseDuration(windowString)
if err != nil {
b.Error(err)
}
case <-b.quit:
// quit the block
return
case msg := <-b.in:
// deal with inbound data
if tree == nil {
break
}
val, err := jee.Eval(tree, msg)
if err != nil {
b.Error(err)
break
}
// TODO make this a type swtich and convert anything we can to a
// float
val, ok := val.(float64)
if !ok {
b.Error(errors.New("trying to put a non-float into the moving average"))
continue
}
queueMessage := &PQMessage{
val: val,
t: time.Now(),
}
heap.Push(pq, queueMessage)
case <-b.inpoll:
// deal with a poll request
outMsg := map[string]interface{}{
"Average": pqAverage(pq),
}
b.out <- outMsg
case c := <-b.queryavg:
outMsg := map[string]interface{}{
"Average": pqAverage(pq),
}
c <- outMsg
case c := <-b.queryrule:
// deal with a query request
c <- map[string]interface{}{
"Path": path,
"Window": windowString,
}
case <-waitTimer.C:
}
for {
pqMsg, diff := pq.PeekAndShift(time.Now(), window)
if pqMsg == nil {
// either the queue is empty, or it's not time to emit
if diff == 0 {
// then the queue is empty. Pause for 5 seconds before checking again
diff = time.Duration(500) * time.Millisecond
}
waitTimer.Reset(diff)
break
}
}
}
}