telegraf/plugins/parsers/phasor_binary/parser.go

package binary_phasor

import (
	"encoding/binary"
	"errors"
	"math"
	"strconv"
	"strings"
	"time"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/metric"
	"github.com/influxdata/telegraf/plugins/parsers"
)

// Parser adheres to the parser interface, contains the parser configuration, and data required to parse binary_phasor
type Parser struct {
	Log telegraf.Logger
	// // measurement is the name of the current config used in each line protocol
	// measurement string

	// // parseMutex is here because Parse() is not threadsafe.  If it is made threadsafe at some point, then we won't need it anymore.
	// parseMutex sync.Mutex

	// pointFrequency data point frequency in one second
	pointFrequency int

	defaultMetricName string
}

const (
	deviceTypeI = 0x01
	deviceTypeU = 0x02

	dataLengthI = 11306
	dataLengthU = 14106
)

const (
	ycPrefix = "tm"
	yxPrefix = "ts"
)

func (p *Parser) Init() error {
	p.pointFrequency = 50
	return nil
}

func (p *Parser) Parse(data []byte, topic string) ([]telegraf.Metric, error) {
	metrics, deviceType, err := p.checkHeaderAndInitMetrics(data, topic)
	if err != nil {
		return nil, err
	}

	p.fillAnalogChanMetrics(metrics, data, 6)
	p.fillSwitchChanMetrics(metrics, data, 9606)

	switch deviceType {
	case deviceTypeI:
		p.fillPQSPFChanMetrics(metrics, data, 9706)
	case deviceTypeU:
		p.fillFdFChanMetrics(metrics, data, 9706)
		p.fillUABUBCUCAChanMetrics(metrics, data, 10506)
	default:
		return nil, errors.New("illegal device type")
	}

	return metrics, nil
}

func (p *Parser) ParseLine(line string) (telegraf.Metric, error) {
	return nil, errors.New("not implemented")
}

func (p *Parser) SetDefaultTags(tags map[string]string) {

}

func init() {
	// Register all variants
	parsers.Add("phasor_binary",
		func(defaultMetricName string) telegraf.Parser {
			return &Parser{defaultMetricName: defaultMetricName}
		},
	)
}

// simply check the data, and initialize metrics with data and topic
func (p *Parser) checkHeaderAndInitMetrics(data []byte, topic string) ([]telegraf.Metric, int, error) {
	if len(data) < 6 {
		return nil, 0, errors.New("no valid data")
	}

	second := int64(binary.LittleEndian.Uint32(data[:4]))
	deviceType := int(data[4])
	metrics := make([]telegraf.Metric, p.pointFrequency)

	device, _ := strings.CutSuffix(topic, "_Phasor")
	switch deviceType {
	case deviceTypeI:
		if len(data) < dataLengthI {
			return nil, 0, errors.New("illegal current data length")
		}
		if data[5] != 0x0e { // 14, current channel number
			return nil, 0, errors.New("illegal current channel number")
		}

		for i := range metrics {
			metrics[i] = metric.New("current",
				map[string]string{"device": device},
				make(map[string]any, 44), // 3*8+2*8+4
				time.Unix(second, int64(i*1e9/p.pointFrequency)))
		}
	case deviceTypeU:
		if len(data) < dataLengthU {
			return nil, 0, errors.New("illegal voltage data length")
		}
		if data[5] != 0x0f { // 15, voltage channel number
			return nil, 0, errors.New("illegal voltage channel number")
		}

		for i := range metrics {
			metrics[i] = metric.New("voltage",
				map[string]string{"device": device},
				make(map[string]any, 51), // 3*8+2*8+2+3*3
				time.Unix(second, int64(i*1e9/p.pointFrequency)))
		}
	default:
		return nil, 0, errors.New("illegal device type")
	}

	return metrics, deviceType, nil
}

// yc metrics
func (p *Parser) fillAnalogChanMetrics(metrics []telegraf.Metric, data []byte, begin int) {

	for ci := range 8 {
		chanNo := strconv.Itoa(ci + 1)
		for mj := range metrics {
			b := begin + (ci*p.pointFrequency+mj)*24

			amp := math.Float64frombits(binary.LittleEndian.Uint64(data[b : b+8]))
			pa := math.Float64frombits(binary.LittleEndian.Uint64(data[b+8 : b+16]))
			rms := math.Float64frombits(binary.LittleEndian.Uint64(data[b+16 : b+24]))

			metrics[mj].AddField(ycPrefix+chanNo+"_amp", amp)
			metrics[mj].AddField(ycPrefix+chanNo+"_pa", pa)
			metrics[mj].AddField(ycPrefix+chanNo+"_rms", rms)
		}
	}
}

// yx metrics
func (p *Parser) fillSwitchChanMetrics(metrics []telegraf.Metric, data []byte, begin int) {

	for ci := range 2 {
		for mj := range metrics {
			b := begin + ci*p.pointFrequency + mj
			for bk := range 8 {
				chanNo := strconv.Itoa(ci*8 + bk + 1)

				metrics[mj].AddField(yxPrefix+chanNo, uint8((data[b]>>bk)&1))
			}
		}
	}
}

// current relative metrics
func (p *Parser) fillPQSPFChanMetrics(metrics []telegraf.Metric, data []byte, begin int) {

	for ci, channel := range []string{"p", "q", "s", "pf"} {
		for mj := range metrics {
			b := begin + (ci*p.pointFrequency+mj)*8

			metrics[mj].AddField(channel, math.Float64frombits(binary.LittleEndian.Uint64(data[b:b+8])))
		}
	}
}

// voltage relative metrics
func (p *Parser) fillFdFChanMetrics(metrics []telegraf.Metric, data []byte, begin int) {

	for ci, channel := range []string{"f", "df"} {
		for mj := range metrics {
			b := begin + (ci*p.pointFrequency+mj)*8

			metrics[mj].AddField(channel, math.Float64frombits(binary.LittleEndian.Uint64(data[b:b+8])))
		}
	}
}

// voltage metrics
func (p *Parser) fillUABUBCUCAChanMetrics(metrics []telegraf.Metric, data []byte, begin int) {

	for ci, channel := range []string{"uab", "ubc", "uca"} {
		for mj := range metrics {
			b := begin + (ci*p.pointFrequency+mj)*24

			amp := math.Float64frombits(binary.LittleEndian.Uint64(data[b : b+8]))
			pa := math.Float64frombits(binary.LittleEndian.Uint64(data[b+8 : b+16]))
			rms := math.Float64frombits(binary.LittleEndian.Uint64(data[b+16 : b+24]))

			metrics[mj].AddField(channel+"_amp", amp)
			metrics[mj].AddField(channel+"_pa", pa)
			metrics[mj].AddField(channel+"_rms", rms)
		}
	}
}