uecs2influxdb1x.go

//------------------------------------------------------
// UECS CCM to InfluxDB 1.x
// 2023.04.18  ookuma yousuke
// 
// ［必要パッケージ］
// 		go get gopkg.in/ini.v1
// 		go get github.com/influxdata/influxdb/client/v2
// 		go get github.com/deckarep/golang-set
// ［build］
// 		go build uecs2influxdb1x.go
//  [必要iniファイル]
//		grafux.cfg
//			https://bitbucket.org/bigbearfarm/grafux/src/master/
// ［実行］
// 		./uecs2influxdb1x
//------------------------------------------------------

package main

import (
	"encoding/xml"
	"fmt"
	"net"
	"strings"
	"strconv"
	"time"
	"os"
	"gopkg.in/ini.v1"
	"github.com/influxdata/influxdb/client/v2"
	mapset "github.com/deckarep/golang-set"
)

type UECS struct {
	XMLName xml.Name `xml:"UECS"`
	DATA    struct {
		XMLName  xml.Name `xml:"DATA"`
		Type     string   `xml:"type,attr"`
		Room     int      `xml:"room,attr"`
		Region   int      `xml:"region,attr"`
		Order    int      `xml:"order,attr"`
		Priority int      `xml:"priority,attr"`
		Value    float64  `xml:",chardata"`
	}
	IP string `xml:"IP"`
}

// iniファイルの読込み
func ReadIniFile(filepath string) (*ini.File, error) {
	cfg, err := ini.Load(filepath)
	if err != nil {
		return nil, err
	}
	return cfg, nil
}

// iniファイルの読込み keyの抽出
func getKeysFromINIFile(fileName string, sectionName string) ([]string, error) {
	cfg, err := ini.Load(fileName)
	if err != nil {
		return nil, err
	}
	section, err := cfg.GetSection(sectionName)
	if err != nil {
		return nil, err
	}
	keys := make([]string, 0)
	for _, key := range section.Keys() {
		if key.Value() != "" {
			keys = append(keys, strings.ToLower(key.Name())) // 小文字にする
		}
	}
	return keys, nil
}

//UDP 受信 ＆ influxdb write
func receiveAndWriteToInfluxdb(InfluxDB_config map[string]string, set mapset.Set) {
	jst, err := time.LoadLocation("Asia/Tokyo") // JSTのタイムゾーンを取得
	if err != nil {
		panic(err)
	}
	// リッスンアドレスとポートを指定
	addr, err := net.ResolveUDPAddr("udp", "224.0.0.1:16520")
	if err != nil {
		panic(err)
	}
	// UDPコネクションを作成
	conn, err := net.ListenMulticastUDP("udp", nil, addr)
	if err != nil {
		panic(err)
	}

	// InfluxDBクライアントの作成
	influxClient, err := client.NewHTTPClient(client.HTTPConfig{
		Addr:     "http://"+ InfluxDB_config["host"] +":8086",
		Username: InfluxDB_config["username"],
		Password: InfluxDB_config["password"],
	})
	if err != nil {
		fmt.Println("Error: ", err)
		os.Exit(1)
	}
	defer influxClient.Close()

	buf := make([]byte, 512)
	for {
		n, _, err := conn.ReadFromUDP(buf)
		if err != nil {
			fmt.Println("Error: ", err)
			continue
		}

		// 受信データをパースして構造体に格納
		var u UECS
		if err := xml.Unmarshal(buf[:n], &u); err != nil {
			panic(err)
		}

		s := u.DATA.Type
		idx := strings.Index(s, ".") // .以前の文字列
		s1 := strings.ToLower(s[:idx])  // 小文字にする

		measurement := fmt.Sprintf("%s_%s_%s_%s_%s", s1, strconv.Itoa(u.DATA.Room), strconv.Itoa(u.DATA.Region), strconv.Itoa(u.DATA.Order), strconv.Itoa(u.DATA.Priority))
		sval := strconv.FormatFloat(u.DATA.Value, 'f', -1, 64)

		// 収集するCCMのみInfluxdbへ書き込む
		if set.Contains(measurement) {
			// InfluxDBに書き込むデータの作成
			bp, err := client.NewBatchPoints(client.BatchPointsConfig{
				Database:  InfluxDB_config["MyDB"],
				Precision: "s",
			})
			if err != nil {
				panic(err)
			}
			tags := map[string]string{"Flag": "0"} //Flagタグに"0"をセット
			fields := map[string]interface{}{
				"Value": u.DATA.Value,
			}

			pt, err := client.NewPoint(measurement, tags, fields, time.Now())
			if err != nil {
				panic(err)
			}
			bp.AddPoint(pt)

			if err := influxClient.Write(bp); err != nil {
				panic(err)
			}
			// 収集するCCMのみ表示させる
			t := time.Now().In(jst) // 現在時刻をJSTに変換
			fmt.Printf("%-30s%-20s%-15s%-13s\n", measurement ,sval, u.IP, t.Format("15:04:05.000"))
//			fmt.Printf("%-4s%-30s%-20s%-15s\n","【○】", measurement ,sval, u.IP)
		} else {
			fmt.Printf("")
//			fmt.Printf("%-4s%-30s%-20s%-15s\n", "【☓】",measurement ,sval, u.IP)

		}
	}
}


func main() {

	// ini ファイル読み込み
	fileName := "grafux.cfg"
	cfg, err := ReadIniFile(fileName)
	if err != nil {
		fmt.Println("Failed to read INI file:", err)
		return
	}
	// InfluxDB 設定 map[string]int型の変数を宣言して初期化する
	InfluxDB_config := map[string]string{"host": cfg.Section("influx_db").Key("local_host").String(),
										"username": cfg.Section("influx_db").Key("user_name").String(),
										"password": cfg.Section("influx_db").Key("user_pass").String(),
										"MyDB": cfg.Section("influx_db").Key("db_name").String(),
									}


	//収集するccmをリスト化
	sectionName := "table_name"
	keys, err := getKeysFromINIFile(fileName, sectionName)
	if err != nil {
		fmt.Println("Error:", err)
		os.Exit(1)
	}
	fmt.Println("【収集するCCM一覧】")
	fmt.Println(keys)
	// keyの集合関数作成
	set := mapset.NewSet()
	for _, key := range keys {
		set.Add(key)
	}

	//UDP受信 & Influxdb書込み
	fmt.Println("")
	fmt.Printf("%-30s%-20s%-15s%-13s\n", "<Measurement>" ,"<Value>", "<IP>", "<hh:mm:ss.sss>")
	receiveAndWriteToInfluxdb(InfluxDB_config, set)

}