Commit 5fb44595 authored by 薄玉虎's avatar 薄玉虎
Browse files

1.增加算法场景日志信息 

2.增加flink的运行程序jobname
3.调整乌海矿于海南矿摄像头id不同区别的逻辑
parent a20dfdc7

src/main/java/com/censoft/flink/StreamingJob.java

+13 −4
Original line number Original line Diff line number Diff line
@@ -49,6 +49,7 @@ public class StreamingJob { 




        //2、接收消息并将数据转化为流

        //2、接收消息并将数据转化为流

        //根据场景id 拼接对应的mqtt频道名称

        //根据场景id 拼接对应的mqtt频道名称

//        DataStream<AlgorithmPushDto> mqttStream = env.addSource(new MqttConsumer("/censoft/cpptest/10"));

        DataStream<AlgorithmPushDto> mqttStream = env.addSource(new MqttConsumer("/censoft/cpptest/" + (sceneId - 1)));

        DataStream<AlgorithmPushDto> mqttStream = env.addSource(new MqttConsumer("/censoft/cpptest/" + (sceneId - 1)));





        //3、进行处理

        //3、进行处理
@@ -86,9 +87,14 @@ public class StreamingJob { 
            warnPushBase.setAlarmDeviceName(cameraPo.getCameraName());

            warnPushBase.setAlarmDeviceName(cameraPo.getCameraName());

            //获取摄像头归属

            //获取摄像头归属

            warnPushBase.setBelong(cameraPo.getBelong());

            warnPushBase.setBelong(cameraPo.getBelong());

            //获取摄像头Id

            //获取摄像头Id   乌海矿与海南矿实际情况不同

            if("乌海矿".equals(cameraPo.getBelong())){

                warnPushBase.setChannelId(cameraPo.getRtsp()+"$1$0$0");

                warnPushBase.setChannelId(cameraPo.getRtsp()+"$1$0$0");

            warnPushBase.setGradeName("二级预警");

            }else{

                warnPushBase.setChannelId(cameraPo.getRtsp().replace("_","$1$0$"));

            }



            warnPushBase.setGradeName("预警");

            //获取摄像头名称

            //获取摄像头名称

            String algorithmName = getAlgorithmPoById(algorithmPoList, Long.valueOf(algorithmPushDto.getAlgorithmName()))

            String algorithmName = getAlgorithmPoById(algorithmPoList, Long.valueOf(algorithmPushDto.getAlgorithmName()))

                    .getAlgorithmName();

                    .getAlgorithmName();
@@ -116,7 +122,10 @@ public class StreamingJob { 
        outputStreamOperator.print();

        outputStreamOperator.print();

        outputStreamOperator.writeAsText("F:/word" + sceneId + ".txt", FileSystem.WriteMode.OVERWRITE).setParallelism(1);

        outputStreamOperator.writeAsText("F:/word" + sceneId + ".txt", FileSystem.WriteMode.OVERWRITE).setParallelism(1);

        //3、7 自动执行

        //3、7 自动执行

        env.execute();

        env.execute(sceneId.toString());







    }

    }

src/main/java/com/censoft/flink/transform/AlarmSecondFilterFunction.java

+1 −0
Original line number Original line Diff line number Diff line
@@ -60,6 +60,7 @@ public class AlarmSecondFilterFunction implements AlgorithmBaseFilterFunction { 
            cache.put(algorithmType, algorithmPushDto.getTimeStamp());

            cache.put(algorithmType, algorithmPushDto.getTimeStamp());

            return true;

            return true;

        }

        }

        System.out.println(algorithmPushDto.getCameraName()+"--"+algorithmPushDto.getAlgorithmName()+"--"+"未超过报警时间间隔");

        return false;

        return false;

    }

    }

}
 
}

src/main/java/com/censoft/flink/transform/AlarmThresholdFunction.java

+1 −0
Original line number Original line Diff line number Diff line
@@ -50,6 +50,7 @@ public class AlarmThresholdFunction implements AlgorithmBaseFilterFunction { 
                .stream()

                .stream()

                .map(po -> {

                .map(po -> {

                    String label = po.getLabel();

                    String label = po.getLabel();

                    System.out.println(label);

                    if (label.contains("_")) {

                    if (label.contains("_")) {

                        String thresholdString = label.substring(label.indexOf("_") + 1);

                        String thresholdString = label.substring(label.indexOf("_") + 1);

                        if (NumberUtil.isInteger(thresholdString)) {

                        if (NumberUtil.isInteger(thresholdString)) {

src/main/java/com/censoft/flink/transform/AlgorithmTypeFlatMapFunction.java

+3 −0
Original line number Original line Diff line number Diff line
@@ -29,6 +29,7 @@ public class AlgorithmTypeFlatMapFunction implements FlatMapFunction<AlgorithmPu 
        //2、将相同label的Key值分出多个预警

        //2、将相同label的Key值分出多个预警





        List<AlgorithmPushResultBoxDto> boxDtos = algorithmPushDto.getResult();

        List<AlgorithmPushResultBoxDto> boxDtos = algorithmPushDto.getResult();

        System.out.println(algorithmPushDto.toString());





        List<String> types = boxDtos

        List<String> types = boxDtos

                .stream()

                .stream()
@@ -53,6 +54,8 @@ public class AlgorithmTypeFlatMapFunction implements FlatMapFunction<AlgorithmPu 
            pushDto.setPictureAddress(algorithmPushDto.getPictureAddress());

            pushDto.setPictureAddress(algorithmPushDto.getPictureAddress());

            pushDto.setSort(algorithmPushDto.getSort());

            pushDto.setSort(algorithmPushDto.getSort());

            pushDto.setTimeStamp(algorithmPushDto.getTimeStamp());

            pushDto.setTimeStamp(algorithmPushDto.getTimeStamp());

            pushDto.setServerIp(algorithmPushDto.getServerIp());

            pushDto.setRemark(algorithmPushDto.getRemark());





            pushDto.setAlgorithmType(type);

            pushDto.setAlgorithmType(type);