<commit_before>package it.uniroma2.ember;

/**
 * This is the routing topology for Apache Flink operators and transformations
 * Project Ember entrypoint
 */


import it.uniroma2.ember.elasticsearch.EmberElasticsearchAlertSource;
import it.uniroma2.ember.elasticsearch.EmberElasticsearchRankSinkFunction;
import it.uniroma2.ember.elasticsearch.EmberElasticsearchSinkFunction;
import it.uniroma2.ember.kafka.EmberKafkaControlSink;
import it.uniroma2.ember.kafka.EmberKafkaProducer;
import it.uniroma2.ember.operators.join.EmberAggregateSensors;
import it.uniroma2.ember.operators.join.EmberControlRoom;
import it.uniroma2.ember.operators.join.EmberEmptyApply;
import it.uniroma2.ember.operators.parser.EmberParseLamp;
import it.uniroma2.ember.operators.parser.EmberParseLumen;
import it.uniroma2.ember.operators.parser.EmberParseTraffic;
import it.uniroma2.ember.operators.selector.*;
import it.uniroma2.ember.operators.serializer.*;
import it.uniroma2.ember.stats.*;
import it.uniroma2.ember.utils.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.*;

public class CityOfLight {

    private final static long WINDOW_TIME_SEC  = 10;

    private final static long WINDOW_CONSUMPTION_HOUR_MINUTES = 24;

    private static long MONITOR_TIME_MINUTES_MIN = 1;
    private static long MONITOR_TIME_MINUTES_MAX = 60;
    public static int MONITOR_MAX_LEN = 10;
    public static int MAX_LIFE_SPAN_DAYS = 200;

    public static long ALERT_SOURCE_PERIOD_SECONDS = 30 * 1000;
    public static long TO_FAILURE_SECONDS = 30 * 3;

    private static String CLUSTER_NAME = "embercluster";
    private static String CLUSTER_ADDRESS = "db.project-ember.city";

    private static int CLUSTER_PORT = 9300;

    @SuppressWarnings("unchecked")
    public static void main(String[] argv) throws Exception {

        if (argv.length == 2) {
            // assuming a new property file to open
            try {
                ParameterTool parameters = ParameterTool.fromPropertiesFile(argv[1]);

                MONITOR_TIME_MINUTES_MIN = parameters.getLong("lifespan.rank.min", 1);
                MONITOR_TIME_MINUTES_MAX = parameters.getLong("lifespan.rank.max", 60);
                MONITOR_MAX_LEN          = parameters.getInt("lifespan.rank.size", 10);
                MAX_LIFE_SPAN_DAYS       = parameters.getInt("lifespan.days.max", 200);

                ALERT_SOURCE_PERIOD_SECONDS = parameters.getLong("alerts.period.seconds", 30000);
                TO_FAILURE_SECONDS          = parameters.getLong("alerts.electricalfailure.seconds", 90);

                CLUSTER_NAME             = parameters.get("elasticsearch.cluster.name", "embercluster");
                CLUSTER_ADDRESS          = parameters.get("elasticsearch.cluster.address", "localhost");
                CLUSTER_PORT             = parameters.getInt("elasticsearch.cluster.port", 9300);

            } catch (Exception e){
                e.printStackTrace();
                System.out.println("Error reading property file... exiting now!");
                return;
            }
        }

        // set up the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // get input data
        Properties properties = new Properties();

        // setting group id
        /* to be setted by config file eventually */
        properties.setProperty("bootstrap.servers", "kafka.project-ember.city:9092");
        //properties.setProperty("group.id", "thegrid");
        properties.setProperty("heartbeat.interval.ms", "10000");

        // preparing elasticsearch config for Elasticsearch API only
        Map<String, Object> elasticConfig = new HashMap<>();
        elasticConfig.put("cluster.address", CLUSTER_ADDRESS);
        elasticConfig.put("cluster.port", CLUSTER_PORT);
        elasticConfig.put("cluster.name", CLUSTER_NAME);

        // preparing elasticsearch for Elasticsearch Connector
        Map<String,String> config = new HashMap<>();
        config.put("bulk.flush.max.actions","1");
        config.put("cluster.name", CLUSTER_NAME);

        List<InetSocketAddress> transports = new ArrayList<>();
        transports.add(new InetSocketAddress(InetAddress.getByName(CLUSTER_ADDRESS), CLUSTER_PORT));

        // STREETLAMPS DATA PROCESSING
        // setting topic and processing the stream from streetlamps<commit_msg>Add support for Kafka from .properties config file.<commit_after>package it.uniroma2.ember;

/**
 * This is the routing topology for Apache Flink operators and transformations
 * Project Ember entrypoint
 */


import it.uniroma2.ember.elasticsearch.EmberElasticsearchAlertSource;
import it.uniroma2.ember.elasticsearch.EmberElasticsearchRankSinkFunction;
import it.uniroma2.ember.elasticsearch.EmberElasticsearchSinkFunction;
import it.uniroma2.ember.kafka.EmberKafkaControlSink;
import it.uniroma2.ember.kafka.EmberKafkaProducer;
import it.uniroma2.ember.operators.join.EmberAggregateSensors;
import it.uniroma2.ember.operators.join.EmberControlRoom;
import it.uniroma2.ember.operators.join.EmberEmptyApply;
import it.uniroma2.ember.operators.parser.EmberParseLamp;
import it.uniroma2.ember.operators.parser.EmberParseLumen;
import it.uniroma2.ember.operators.parser.EmberParseTraffic;
import it.uniroma2.ember.operators.selector.*;
import it.uniroma2.ember.operators.serializer.*;
import it.uniroma2.ember.stats.*;
import it.uniroma2.ember.utils.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.*;

public class CityOfLight {

    private final static long WINDOW_TIME_SEC  = 10;

    private final static long WINDOW_CONSUMPTION_HOUR_MINUTES = 24;

    private static long MONITOR_TIME_MINUTES_MIN = 1;
    private static long MONITOR_TIME_MINUTES_MAX = 60;
    public static int MONITOR_MAX_LEN            = 10;
    public static int MAX_LIFE_SPAN_DAYS         = 200;

    public static long ALERT_SOURCE_PERIOD_SECONDS = 30 * 1000;
    public static long TO_FAILURE_SECONDS          = 30 * 3;

    private static String ELASTICSEARCH_NAME    = "embercluster";
    private static String ELASTICSEARCH_ADDRESS = "db.project-ember.city";
    private static int ELASTICSEARCH_PORT       = 9300;

    private static String KAFKA_ADDRESS = "kafka.project-ember.city";
    private static int KAFKA_PORT       = 9092;

    @SuppressWarnings("unchecked")
    public static void main(String[] argv) throws Exception {

        if (argv.length == 2) {
            // assuming a new property file to open
            try {
                ParameterTool parameters = ParameterTool.fromPropertiesFile(argv[1]);

                MONITOR_TIME_MINUTES_MIN    = parameters.getLong("lifespan.rank.min", 1);
                MONITOR_TIME_MINUTES_MAX    = parameters.getLong("lifespan.rank.max", 60);
                MONITOR_MAX_LEN             = parameters.getInt("lifespan.rank.size", 10);
                MAX_LIFE_SPAN_DAYS          = parameters.getInt("lifespan.days.max", 200);

                ALERT_SOURCE_PERIOD_SECONDS = parameters.getLong("alerts.period.seconds", 30000);
                TO_FAILURE_SECONDS          = parameters.getLong("alerts.electricalfailure.seconds", 90);

                ELASTICSEARCH_NAME          = parameters.get("elasticsearch.cluster.name", "embercluster");
                ELASTICSEARCH_ADDRESS       = parameters.get("elasticsearch.cluster.address", "localhost");
                ELASTICSEARCH_PORT          = parameters.getInt("elasticsearch.cluster.port", 9300);

                KAFKA_ADDRESS               = parameters.get("kafka.cluster.address", "localhost");
                KAFKA_PORT                  = parameters.getInt("kafka.cluster.port", 9092);

            } catch (Exception e){
                e.printStackTrace();
                System.out.println("Error reading property file... exiting now!");
                return;
            }
        }

        // set up the execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // get input data
        Properties properties = new Properties();

        // setting group id
        /* to be setted by config file eventually */
        properties.setProperty("bootstrap.servers", KAFKA_ADDRESS + ":" + String.valueOf(KAFKA_PORT));
        properties.setProperty("heartbeat.interval.ms", "10000");

        // preparing elasticsearch config for Elasticsearch API only
        Map<String, Object> elasticConfig = new HashMap<>();
        elasticConfig.put("cluster.address", ELASTICSEARCH_ADDRESS);
        elasticConfig.put("cluster.port", ELASTICSEARCH_PORT);
        elasticConfig.put("cluster.name", ELASTICSEARCH_NAME);

        // preparing elasticsearch for Elasticsearch Connector
        Map<String,String> config = new HashMap<>();
        config.put("bulk.flush.max.actions","1");
        config.put("cluster.name", ELASTICSEARCH_NAME);

        List<InetSocketAddress> transports = new ArrayList<>();
        transports.add(new InetSocketAddress(InetAddress.getByName(ELASTICSEARCH_ADDRESS), ELASTICSEARCH_PORT));

        // STREETLAMPS DATA PROCESSING
        // setting topic and processing the stream from streetlamps
{'fraction_non_alphanumeric': 0.09582675784674732, 'fraction_numerical': 0.013852358407855515, 'mean_word_length': 3.5033557046979866, 'pattern_counts': {'":': 1, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '818883', 'n_tokens_mistral': 3445, 'n_tokens_neox': 3263, 'n_words': 589}