timeseries/timeseries.go

Rama main · réplica de lectura del repositorio original.

package timeseries

import (
	"encoding/json"
	"fmt"
	"math"
	"strings"

	"github.com/vmihailenco/msgpack/v5"
)

var NaN = float32(math.NaN())

func IsNaN(v float32) bool {
	return v != v
}

func IsInf(f float32, sign int) bool {
	return sign >= 0 && f > math.MaxFloat32 || sign <= 0 && f < -math.MaxFloat32
}

type TimeSeries struct {
	from Time
	step Duration
	data []float32
	last float32
}

func New(from Time, pointsCount int, step Duration) *TimeSeries {
	data := make([]float32, pointsCount)
	for i := range data {
		data[i] = NaN
	}
	return NewWithData(from, step, data)
}

func NewWithData(from Time, step Duration, data []float32) *TimeSeries {
	ts := &TimeSeries{
		from: from,
		step: step,
		data: data,
		last: data[len(data)-1],
	}
	return ts
}

func (ts *TimeSeries) Len() int {
	if ts.IsEmpty() {
		return 0
	}
	return len(ts.data)
}

func (ts *TimeSeries) MarshalJSON() ([]byte, error) {
	if ts.IsEmpty() {
		return json.Marshal(nil)
	}
	vs := make([]Value, 0, ts.Len())
	iter := ts.Iter()
	for iter.Next() {
		_, v := iter.Value()
		vs = append(vs, Value(v))
	}
	if len(vs) == 0 {
		return json.Marshal(nil)
	}
	d, err := json.Marshal(vs)
	return d, err
}

func (ts *TimeSeries) UnmarshalJSON(data []byte) error {
	if string(data) == "null" {
		ts.data = nil
		ts.last = NaN
		return nil
	}

	var raw []*float32
	if err := json.Unmarshal(data, &raw); err != nil {
		return err
	}

	values := make([]float32, len(raw))
	for i, v := range raw {
		if v == nil {
			values[i] = NaN
		} else {
			values[i] = *v
		}
	}

	ts.data = values
	if len(values) > 0 {
		ts.last = values[len(values)-1]
	} else {
		ts.last = NaN
	}

	return nil
}

type Msgpack struct {
	From Time     `msgpack:"from"`
	Step Duration `msgpack:"step"`
	Data []byte   `msgpack:"data"`
}

func (ts *TimeSeries) EncodeMsgpack(enc *msgpack.Encoder) error {
	m := Msgpack{
		From: ts.from,
		Step: ts.step,
		Data: asBytes32(ts.data),
	}
	return enc.Encode(m)
}

func (ts *TimeSeries) DecodeMsgpack(dec *msgpack.Decoder) error {
	var m Msgpack
	err := dec.Decode(&m)
	if err != nil {
		return err
	}
	ts.from = m.From
	ts.step = m.Step
	ts.last = NaN
	if len(m.Data) > 0 {
		ts.data = asFloats32(m.Data)
		ts.last = ts.data[len(ts.data)-1]
	}
	return nil
}

func (ts *TimeSeries) String() string {
	if ts.IsEmpty() {
		return "TimeSeries(nil)"
	}
	values := make([]string, 0, ts.Len())
	iter := ts.Iter()
	for iter.Next() {
		_, v := iter.Value()
		values = append(values, Value(v).String())
	}
	return fmt.Sprintf("TimeSeries(%d, %d, %d, [%s])", ts.from, ts.Len(), ts.step, strings.Join(values, " "))
}

func (ts *TimeSeries) Get() *TimeSeries {
	return ts
}

func (ts *TimeSeries) Set(t Time, v float32) {
	t = t.Truncate(ts.step)
	if t < ts.from {
		return
	}
	idx := int((t - ts.from) / Time(ts.step))
	l := len(ts.data) - 1
	if idx <= l {
		ts.data[idx] = v
		if idx == l {
			ts.last = v
		}
	}
}

type FillFunc func(ts *TimeSeries, from Time, step Duration, data []float32) bool

func FillAny(ts *TimeSeries, from Time, step Duration, data []float32) bool {
	changed := false
	maxIndex := len(ts.data) - 1
	tSrc, iSrc := from, 0
	if ts.from.Sub(tSrc) >= ts.step {
		tSrc = tSrc.Add(ts.from.Sub(tSrc.Truncate(ts.step)).Truncate(ts.step))
		if tSrc > ts.from {
			tSrc = tSrc.Add(-ts.step)
		}
		iSrc = int((tSrc - from) / Time(step))
	}
	tDst, iDst := ts.from, 0
	if tSrc > tDst {
		tDst = tSrc.Truncate(ts.step)
		if tDst < tSrc {
			tDst = tDst.Add(ts.step)
		}
		iDst = int((tDst - ts.from) / Time(ts.step))
	}
	vv := NaN
	for _, v := range data[iSrc:] {
		if tSrc > tDst {
			ts.data[iDst] = vv
			vv = NaN
			iDst++
			if iDst > maxIndex {
				break
			}
			tDst += Time(ts.step)
		}
		if !IsNaN(v) {
			vv = v
			changed = true
		}
		tSrc += Time(step)
	}
	if iDst <= maxIndex {
		ts.data[iDst] = vv
	}
	ts.last = ts.data[maxIndex]
	return changed
}

func FillSum(ts *TimeSeries, from Time, step Duration, data []float32) bool {
	changed := false
	maxIndex := len(ts.data) - 1
	tSrc, iSrc := from, 0
	if ts.from.Sub(tSrc) >= ts.step {
		tSrc = tSrc.Add(ts.from.Sub(tSrc.Truncate(ts.step)).Truncate(ts.step))
		if tSrc > ts.from {
			tSrc = tSrc.Add(-ts.step)
		}
		iSrc = int((tSrc - from) / Time(step))
	}
	tDst, iDst := ts.from, 0
	if tSrc > tDst {
		tDst = tSrc.Truncate(ts.step)
		if tDst < tSrc {
			tDst = tDst.Add(ts.step)
		}
		iDst = int((tDst - ts.from) / Time(ts.step))
	}
	vv := float32(0)
	if !IsNaN(ts.data[iDst]) {
		vv = ts.data[iDst]
	}
	for _, v := range data[iSrc:] {
		if tSrc > tDst {
			ts.data[iDst] = vv
			vv = 0
			iDst++
			if iDst > maxIndex {
				break
			}
			tDst += Time(ts.step)
		}
		if !IsNaN(v) {
			vv += v
			changed = true
		}
		tSrc += Time(step)
	}
	if iDst <= maxIndex {
		ts.data[iDst] = vv
	}
	ts.last = ts.data[maxIndex]
	return changed
}

func FillAvg(ts *TimeSeries, from Time, step Duration, data []float32) bool {
	changed := false
	maxIndex := len(ts.data) - 1
	tSrc, iSrc := from, 0
	if ts.from.Sub(tSrc) >= ts.step {
		tSrc = tSrc.Add(ts.from.Sub(tSrc.Truncate(ts.step)).Truncate(ts.step))
		if tSrc > ts.from {
			tSrc = tSrc.Add(-ts.step)
		}
		iSrc = int((tSrc - from) / Time(step))
	}
	tDst, iDst := ts.from, 0
	if tSrc > tDst {
		tDst = tSrc.Truncate(ts.step)
		if tDst < tSrc {
			tDst = tDst.Add(ts.step)
		}
		iDst = int((tDst - ts.from) / Time(ts.step))
	}
	vv := float32(0)
	cc := float32(0)
	if !IsNaN(ts.data[iDst]) {
		vv = ts.data[iDst]
		cc = 1
	}
	for _, v := range data[iSrc:] {
		if tSrc > tDst {
			if cc > 0 {
				ts.data[iDst] = vv / cc
			}
			vv = 0
			cc = 0
			iDst++
			if iDst > maxIndex {
				break
			}
			tDst += Time(ts.step)
		}
		if !IsNaN(v) {
			vv += v
			cc++
			changed = true
		}
		tSrc += Time(step)
	}
	if iDst <= maxIndex {
		if cc > 0 {
			ts.data[iDst] = vv / cc
		}
	}
	ts.last = ts.data[maxIndex]
	return changed
}

func (ts *TimeSeries) Iter() *Iterator {
	if ts.IsEmpty() {
		return &Iterator{data: nil}
	}
	return &Iterator{
		step: ts.step,
		data: ts.data,
		idx:  -1,
		t:    ts.from.Add(-ts.step),
	}
}

func (ts *TimeSeries) IterFrom(from Time) *Iterator {
	if ts.IsEmpty() {
		return &Iterator{data: nil}
	}
	if from.Before(ts.from) {
		from = ts.from
	}
	to := ts.from.Add(ts.step * Duration(len(ts.data)-1))
	if from.After(to) {
		return &Iterator{data: nil}
	}
	from = from.Truncate(ts.step)
	return &Iterator{
		step: ts.step,
		data: ts.data,
		idx:  int(from.Sub(ts.from)/ts.step) - 1,
		t:    from.Add(-ts.step),
	}
}

func (ts *TimeSeries) IsEmpty() bool {
	return ts == nil
}

func (ts *TimeSeries) Last() float32 {
	if ts.IsEmpty() {
		return NaN
	}
	return ts.last
}

func (ts *TimeSeries) TailIsEmpty() bool {
	if ts.IsEmpty() {
		return true
	}
	if !IsNaN(ts.last) {
		return false
	}
	l := len(ts.data)
	if l >= 2 && !IsNaN(ts.data[l-2]) {
		return false
	}
	if l >= 3 && !IsNaN(ts.data[l-3]) {
		return false
	}
	return true
}

func (ts *TimeSeries) Reduce(f F) float32 {
	if ts.IsEmpty() {
		return NaN
	}
	accumulator := NaN
	iter := ts.Iter()
	for iter.Next() {
		t, v := iter.Value()
		accumulator = f(t, accumulator, v)
	}
	return accumulator
}

func (ts *TimeSeries) ReduceFrom(from Time, f F) float32 {
	if ts.IsEmpty() {
		return NaN
	}
	accumulator := NaN
	iter := ts.IterFrom(from)
	for iter.Next() {
		t, v := iter.Value()
		accumulator = f(t, accumulator, v)
	}
	return accumulator
}

func (ts *TimeSeries) Map(f func(t Time, v float32) float32) *TimeSeries {
	if ts.IsEmpty() {
		return nil
	}

	data := make([]float32, ts.Len())
	iter := ts.Iter()
	i := 0
	for iter.Next() {
		t, v := iter.Value()
		data[i] = f(t, v)
		i++
	}
	return NewWithData(ts.from, ts.step, data)
}

func (ts *TimeSeries) MapInPlace(f func(t Time, v float32) float32) *TimeSeries {
	if ts.IsEmpty() {
		return nil
	}

	t := ts.from
	for i, v := range ts.data {
		ts.data[i] = f(t, v)
		t = t.Add(ts.step)
	}
	ts.last = ts.data[len(ts.data)-1]
	return ts
}

func (ts *TimeSeries) WithNewValue(newValue float32) *TimeSeries {
	if ts.IsEmpty() {
		return nil
	}

	data := make([]float32, ts.Len())
	for i := range data {
		data[i] = newValue
	}
	return NewWithData(ts.from, ts.step, data)
}

func (ts *TimeSeries) NewWithData(data []float32) *TimeSeries {
	if ts.IsEmpty() {
		return nil
	}
	return NewWithData(ts.from, ts.step, data)
}

func (ts *TimeSeries) LastNotNull() (Time, float32) {
	if ts.IsEmpty() {
		return 0, NaN
	}
	var vv float32
	var tt Time
	iter := ts.Iter()
	for iter.Next() {
		t, v := iter.Value()
		if !IsNaN(v) {
			vv = v
			tt = t
		}
	}
	return tt, vv
}

func (ts *TimeSeries) LastNAvg(nPoints int, defaultValue float32) float32 {
	if ts.IsEmpty() || len(ts.data) < nPoints {
		return defaultValue
	}
	sum := float32(0)
	count := 0
	for _, v := range ts.data[len(ts.data)-nPoints:] {
		if !IsNaN(v) {
			sum += v
			count++
		}
	}
	if count > 0 {
		return sum / float32(count)
	}
	return defaultValue
}

func (ts *TimeSeries) LastNMax(nPoints int, defaultValue float32) float32 {
	if ts.IsEmpty() || len(ts.data) < nPoints {
		return defaultValue
	}
	res := defaultValue
	found := false
	for _, v := range ts.data[len(ts.data)-nPoints:] {
		if IsNaN(v) {
			continue
		}
		if !found || v > res {
			res, found = v, true
		}
	}
	return res
}

func (ts *TimeSeries) Average() float32 {
	if ts.IsEmpty() {
		return NaN
	}
	sum, count := float32(0), 0
	iter := ts.Iter()
	for iter.Next() {
		if _, v := iter.Value(); !IsNaN(v) {
			sum += v
			count++
		}
	}
	if count == 0 {
		return NaN
	}
	return sum / float32(count)
}

func Increase(x, status *TimeSeries) *TimeSeries {
	if x.IsEmpty() || status.IsEmpty() {
		return nil
	}
	data := make([]float32, 0, x.Len())
	prev, prevStatus := NaN, NaN
	iter := x.Iter()
	statusIter := status.Iter()
	var d float32
	for iter.Next() && statusIter.Next() {
		_, v := iter.Value()
		d = NaN
		switch {
		case !IsNaN(v) && !IsNaN(prev):
			if v-prev >= 0 {
				d = v - prev
			} else {
				d = v
			}
		case IsNaN(prev) && prevStatus == 1:
			d = v
		}
		prev = v
		_, prevStatus = statusIter.Value()
		data = append(data, d)
	}
	return NewWithData(x.from, x.step, data)
}

func Aggregate2(x, y *TimeSeries, f func(x, y float32) float32) *TimeSeries {
	if x.IsEmpty() || y.IsEmpty() {
		return nil
	}
	data := make([]float32, x.Len())
	xIter := x.Iter()
	yIter := y.Iter()
	i := 0
	for xIter.Next() && yIter.Next() {
		_, xv := xIter.Value()
		_, yv := yIter.Value()
		data[i] = f(xv, yv)
		i++
	}
	return NewWithData(x.from, x.step, data)
}

func Mul(x, y *TimeSeries) *TimeSeries {
	return Aggregate2(x, y, func(x, y float32) float32 { return x * y })
}

func Div(x, y *TimeSeries) *TimeSeries {
	return Aggregate2(x, y, func(x, y float32) float32 { return x / y })
}

func Sub(x, y *TimeSeries) *TimeSeries {
	return Aggregate2(x, y, func(x, y float32) float32 { return x - y })
}

func Sum(x, y *TimeSeries) *TimeSeries {
	return Aggregate2(x, y, func(x, y float32) float32 { return x + y })
}