Experiment with value batching
This commit is contained in:
parent
79c7bf3851
commit
3f163309ad
|
|
@ -14,6 +14,13 @@ type sensorValue struct {
|
|||
Value float64 `json:"value"`
|
||||
}
|
||||
|
||||
func min(a, b int64) int64 {
|
||||
if a < b {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func (s *SensorValuesService) Push(sensorId int64, value float64) (int64, error) {
|
||||
res, err := s.ctx.DB.Exec("INSERT INTO sensor_values (timestamp, sensor_id, value) VALUES (?, ?, ?)", time.Now().Unix(), sensorId, value)
|
||||
|
||||
|
|
@ -25,34 +32,33 @@ func (s *SensorValuesService) Push(sensorId int64, value float64) (int64, error)
|
|||
}
|
||||
|
||||
func (s *SensorValuesService) GetList(sensorId int64, from int64, to int64) ([]sensorValue, error) {
|
||||
var value float64
|
||||
var timestamp int64
|
||||
intervalSize := to - from
|
||||
batchSize := intervalSize
|
||||
|
||||
if intervalSize > 100*24*60 {
|
||||
batchSize = 24 * 60
|
||||
}
|
||||
|
||||
values := make([]sensorValue, 0)
|
||||
current := from
|
||||
|
||||
rows, err := s.ctx.DB.Query("SELECT timestamp, value FROM sensor_values WHERE sensor_id = ? AND timestamp > ? AND timestamp < ? ORDER BY timestamp ASC", sensorId, from, to)
|
||||
for current < to {
|
||||
nextValue := min(to, current+batchSize)
|
||||
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return values, nil
|
||||
}
|
||||
batchValues, err := s.getValueItems(sensorId, current, nextValue)
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
err := rows.Scan(×tamp, &value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
values = append(values, sensorValue{Timestamp: timestamp, Value: value})
|
||||
sum := 0.0
|
||||
|
||||
for _, item := range batchValues {
|
||||
sum += item.Value
|
||||
}
|
||||
|
||||
err = rows.Err()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
values = append(values, sensorValue{Timestamp: int64((nextValue + current) / 2.0), Value: sum / float64(len(batchValues))})
|
||||
current += batchSize
|
||||
}
|
||||
|
||||
return values, nil
|
||||
|
|
@ -70,3 +76,37 @@ func (s *SensorValuesService) GetLatest(sensorId int64, to int64) (*sensorValue,
|
|||
|
||||
return &value, nil
|
||||
}
|
||||
|
||||
func (s *SensorValuesService) getValueItems(sensorId int64, from int64, to int64) ([]sensorValue, error) {
|
||||
values := make([]sensorValue, 0)
|
||||
|
||||
rows, err := s.ctx.DB.Query("SELECT timestamp, value FROM sensor_values WHERE sensor_id = ? AND timestamp > ? AND timestamp < ? ORDER BY timestamp ASC", sensorId, from, to)
|
||||
|
||||
if err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return values, nil
|
||||
}
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
item := sensorValue{}
|
||||
|
||||
err := rows.Scan(&item.Timestamp, &item.Value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
values = append(values, item)
|
||||
}
|
||||
|
||||
err = rows.Err()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return values, nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue