numeric

var Bucket = pipescript.Transform{
	Name:          "bucket",
	Description:   "Puts numbers into custom-sized buckets. Useful for histograms.",
	Documentation: string(resources.MustAsset("docs/transforms/bucket.md")),
	Args: []pipescript.TransformArg{
		{
			Description: "The size of each bucket",
			Optional:    true,
			Default:     pipescript.MustPipe(pipescript.NewConstTransform(10), nil),
			Type:        pipescript.ConstArgType,
			Schema: map[string]interface{}{
				"type": "number",
			},
		},
		{
			Description: "Start location for bucketing",
			Optional:    true,
			Default:     pipescript.MustPipe(pipescript.NewConstTransform(0), nil),
			Type:        pipescript.ConstArgType,
			Schema: map[string]interface{}{
				"type": "number",
			},
		},
	},
	InputSchema: map[string]interface{}{
		"type": "number",
	},
	OutputSchema: map[string]interface{}{
		"type": "string",
	},
	Constructor: pipescript.NewBasic(func(consts []interface{}, pipes []*pipescript.Pipe) ([]interface{}, []*pipescript.Pipe, error) {
		f, ok := pipescript.Float(consts[0])
		if !ok {
			return nil, nil, errors.New("bucket size must be a number")
		}
		consts[0] = f
		f, ok = pipescript.Float(consts[1])
		if !ok {
			return nil, nil, errors.New("bucket start location must be a number")
		}
		consts[1] = f
		return consts, pipes, nil
	}, func(dp *pipescript.Datapoint, args []*pipescript.Datapoint, consts []interface{}, pipes []*pipescript.Pipe, out *pipescript.Datapoint) (*pipescript.Datapoint, error) {
		f, err := dp.Float()
		if err != nil {
			return nil, err
		}
		bucketstart := consts[1].(float64)
		bucketsize := consts[0].(float64)

		bucketnum := math.Floor((f - bucketstart) / bucketsize)

		bucketStartLocation := bucketstart + bucketnum*bucketsize
		bucketEndLocation := bucketstart + (bucketnum+1)*bucketsize
		out.Data = fmt.Sprintf("[%g,%g)", bucketStartLocation, bucketEndLocation)
		return out, nil
	}),
}

var Count = &pipescript.Transform{
	Name:          "count",
	Description:   "returns the total number of datapoints in the stream",
	Documentation: string(resources.MustAsset("docs/transforms/count.md")),
	OutputSchema: map[string]interface{}{
		"type": "number",
	},
	Constructor: pipescript.NewAggregator(func(e *pipescript.TransformEnv, consts []interface{}, pipes []*pipescript.Pipe, out *pipescript.Datapoint) (*pipescript.Datapoint, error) {
		count := int64(0)
		out.Timestamp = 0
		out.Duration = 0
		dp, _, err := e.Next(nil)
		if err != nil {
			return dp, err
		}
		if dp != nil {
			out.Timestamp = dp.Timestamp
		}
		dp2 := dp

		for dp != nil {
			count++
			dp2 = dp
			dp, _, err = e.Next(nil)
			if err != nil {
				return dp, err
			}
		}
		if dp2 != nil {
			out.Duration = dp2.Timestamp + dp2.Duration - out.Timestamp
		}
		out.Data = count
		return out, nil
	}),
}

var Max = &pipescript.Transform{
	Name:        "max",
	Description: "Returns the maximum datapoint in the timeseries",
	Args: []pipescript.TransformArg{
		{
			Description: "The value of this argument is used to check for max",
			Optional:    true,
			Default:     pipescript.IdentityPipe,
			Type:        pipescript.TransformArgType,
			Schema: map[string]interface{}{
				"type": "number",
			},
		},
	},
	Constructor: pipescript.NewAggregator(func(e *pipescript.TransformEnv, consts []interface{}, pipes []*pipescript.Pipe, out *pipescript.Datapoint) (*pipescript.Datapoint, error) {
		args := make([]*pipescript.Datapoint, 1)
		dp, args, err := e.Next(args)
		if err != nil || dp == nil {
			return nil, err
		}
		curmax := dp
		f, err := args[0].Float()
		if err != nil {
			return nil, err
		}
		curval := f
		for {
			dp, args, err = e.Next(args)
			if err != nil {
				return nil, err
			}
			if dp == nil {
				break
			}
			f, err := args[0].Float()
			if err != nil {
				return nil, err
			}
			if f > curval {
				curmax = dp
				curval = f
			}

		}

		out.Data = curmax.Data
		out.Timestamp = curmax.Timestamp
		out.Duration = curmax.Duration
		return out, nil
	}),
}

var Mean = &pipescript.Transform{
	Name:          "mean",
	Description:   "Finds the mean of the timeseries data",
	Documentation: string(resources.MustAsset("docs/transforms/mean.md")),
	InputSchema: map[string]interface{}{
		"type": "number",
	},
	OutputSchema: map[string]interface{}{
		"type": "number",
	},
	Constructor: pipescript.NewAggregator(func(e *pipescript.TransformEnv, consts []interface{}, pipes []*pipescript.Pipe, out *pipescript.Datapoint) (*pipescript.Datapoint, error) {
		dp, _, err := e.Next(nil)
		if err != nil || dp == nil {
			return nil, err
		}
		ldp := dp
		out.Timestamp = dp.Timestamp
		cursum := float64(0)
		count := int64(0)
		for dp != nil {
			f, err := dp.Float()
			if err != nil {
				return nil, err
			}
			cursum += f
			count++
			ldp = dp
			dp, _, err = e.Next(nil)
			if err != nil {
				return nil, err
			}
		}

		out.Data = cursum / float64(count)
		out.Duration = ldp.Timestamp + ldp.Duration - out.Timestamp
		return out, nil
	}),
}

var Min = &pipescript.Transform{
	Name:        "min",
	Description: "Returns the minimum datapoint in the timeseries",
	Args: []pipescript.TransformArg{
		{
			Description: "The value of this argument is used to check for min",
			Optional:    true,
			Default:     pipescript.IdentityPipe,
			Type:        pipescript.TransformArgType,
			Schema: map[string]interface{}{
				"type": "number",
			},
		},
	},
	Constructor: pipescript.NewAggregator(func(e *pipescript.TransformEnv, consts []interface{}, pipes []*pipescript.Pipe, out *pipescript.Datapoint) (*pipescript.Datapoint, error) {
		args := make([]*pipescript.Datapoint, 1)
		dp, args, err := e.Next(args)
		if err != nil || dp == nil {
			return nil, err
		}
		curmin := dp
		f, err := args[0].Float()
		if err != nil {
			return nil, err
		}
		curval := f
		for {
			dp, args, err = e.Next(args)
			if err != nil {
				return nil, err
			}
			if dp == nil {
				break
			}
			f, err := args[0].Float()
			if err != nil {
				return nil, err
			}
			if f < curval {
				curmin = dp
				curval = f
			}

		}

		out.Data = curmin.Data
		out.Timestamp = curmin.Timestamp
		out.Duration = curmin.Duration
		return out, nil
	}),
}

var Sum = &pipescript.Transform{
	Name:          "sum",
	Description:   "Sums data",
	Documentation: string(resources.MustAsset("docs/transforms/sum.md")),
	InputSchema: map[string]interface{}{
		"type": "number",
	},
	OutputSchema: map[string]interface{}{
		"type": "number",
	},
	Constructor: pipescript.NewAggregator(func(e *pipescript.TransformEnv, consts []interface{}, pipes []*pipescript.Pipe, out *pipescript.Datapoint) (*pipescript.Datapoint, error) {
		cursum := float64(0)
		out.Timestamp = 0
		out.Duration = 0
		dp, _, err := e.Next(nil)
		if err != nil {
			return nil, err
		}

		if dp != nil {
			out.Timestamp = dp.Timestamp
		}
		ldp := dp

		for dp != nil {
			f, err := dp.Float()
			if err != nil {
				return nil, err
			}
			cursum += f

			dp, _, err = e.Next(nil)
			if err != nil {
				return nil, err
			}
			ldp = dp
		}

		out.Data = cursum
		if ldp != nil {
			out.Duration = ldp.Timestamp + ldp.Duration - out.Timestamp
		}

		return out, nil
	}),
}