package com.sunwin.metro.process;

import com.alibaba.fastjson.JSONObject;
import com.sunwin.metro.bean.*;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.Gauge;
import io.prometheus.client.exporter.PushGateway;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.io.IOException;

/**
 * @program: sunwin_metro
 * @description: 数据发送到Prometheus
 * @author: xuYJ
 * @create: 2021-12-01 15:48
 **/
public class PrometheusProcess extends KeyedProcessFunction<String, Tuple2<String, String>, String> {
    private PushGateway gateway;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        gateway = new PushGateway("192.168.20.65:9091");
        System.out.println("开启发送Prometheus客户端");
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void processElement(Tuple2<String, String> value, Context ctx, Collector<String> out) throws Exception {
        String[] result = value.f0.split("_");
        try {
            CollectorRegistry pushRegistry = new CollectorRegistry();
            if (value.f1.contains("隧道风机")) {
                BasTefMes basTefMes = JSONObject.parseObject(value.f1, BasTefMes.class);
                Gauge valueToExpose = Gauge.build()
                        .name("BAS_TEF_AI_COMM_WORD")
                        .help("电池总电压")
                        .register(pushRegistry);
                Gauge valueToExpose2 = Gauge.build()
                        .name("BAS_TEF_AI_FAXLET")
                        .help("前轴温度反馈")
                        .register(pushRegistry);
                Gauge valueToExpose3 = Gauge.build()
                        .name("BAS_TEF_AI_Y_VIBR")
                        .help("Y轴振动值")
                        .register(pushRegistry);
                valueToExpose.set(basTefMes.getAi_comm_word());
                valueToExpose2.set(basTefMes.getAi_faxlet());
                valueToExpose3.set(basTefMes.getAi_y_vibr());
            }
            if (value.f1.contains("排热风机")) {
                BasTvfMes basTvfMes = JSONObject.parseObject(value.f1, BasTvfMes.class);
                Gauge valueToExpose = Gauge.build()
                        .name("BAS_TVF_AI_X_VIBR")
                        .help("X轴振动值")
                        .register(pushRegistry);
                Gauge valueToExpose2 = Gauge.build()
                        .name("BAS_TEF_Ai_FAXLET")
                        .help("前轴温度反馈")
                        .register(pushRegistry);
                Gauge valueToExpose3 = Gauge.build()
                        .name("BAS_TVF_AI_Y_VIBR")
                        .help("Y轴振动值")
                        .register(pushRegistry);
                valueToExpose.set(basTvfMes.getAi_x_vibr());
                valueToExpose2.set(basTvfMes.getAi_faxlet());
                valueToExpose3.set(basTvfMes.getAi_y_vibr());
            }
            if (value.f1.contains("不间断电源")) {
                UpsMes ups = JSONObject.parseObject(value.f1, UpsMes.class);
                Gauge valueToExpose = Gauge.build()
                        .name("UPS_AI_DP29")
                        .help("电池电压")
                        .register(pushRegistry);
                Gauge valueToExpose2 = Gauge.build()
                        .name("UPS_AI_DP32")
                        .help("充放电电流")
                        .register(pushRegistry);
                Gauge valueToExpose3 = Gauge.build()
                        .name("UPS_AI_DP36")
                        .help("输出频率")
                        .register(pushRegistry);
                valueToExpose.set(ups.getDp29());
                valueToExpose2.set(ups.getDp32());
                valueToExpose3.set(ups.getDp36());
            }
            if (value.f1.contains("ups巡检仪")) {
                UpsXjyMes upsXyj = JSONObject.parseObject(value.f1, UpsXjyMes.class);
                Gauge valueToExpose = Gauge.build()
                        .name("UPS_XJY_AI_WD")
                        .help("环境温度")
                        .register(pushRegistry);
                Gauge valueToExpose2 = Gauge.build()
                        .name("UPS_XJY_AI_ZDL")
                        .help("组电流")
                        .register(pushRegistry);
                Gauge valueToExpose3 = Gauge.build()
                        .name("UPS_XJY_AI_ZDY")
                        .help("组电压")
                        .register(pushRegistry);
                valueToExpose.set(upsXyj.getWd());
                valueToExpose2.set(upsXyj.getZdl());
                valueToExpose3.set(upsXyj.getZdy());
            }
            if (value.f1.contains("雨水泵")) {
                BasAtMes basAt = JSONObject.parseObject(value.f1, BasAtMes.class);
                Gauge valueToExpose = Gauge.build()
                        .name("BAS_AT_AI_LR")
                        .help("实时水位测量")
                        .register(pushRegistry);
                Gauge valueToExpose2 = Gauge.build()
                        .name("BAS_AT_AI_FLTNO")
                        .help("故障次数")
                        .register(pushRegistry);
                valueToExpose.set(basAt.getAi_lr());
                valueToExpose2.set(basAt.getAi_fltno());
            }
            //利用网关采集数据
            gateway.pushAdd(pushRegistry, result[0] +"_"+  result[1] + "站台_" + result[2] + "设备监控数据");
            out.collect(result[0] +"_"+  result[1] + "站台_" + result[2] + "设备监控数据发送到Prometheus");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) throws IOException {
        PushGateway pushGateway = new PushGateway("192.168.20.65:9091");
        pushGateway.delete("2号线_高铁站站台");
    }
}
