Parsing weather station XML on raspberry pi

Question

This if for a school project that is all about getting the most performance possible out of a relatively weak computer.

For this project we have received a weather station report generator that generates 8000 weather measurements per second. These measurements are split into groups of 10 stations resulting in 800 clusters sending these measurements.

One such measurement:

<?xml version="1.0"?>
<WEATHERDATA>
    <MEASUREMENT>
        <STN>123456</STN>
        <DATE>2009-09-13</DATE>
        <TIME>15:59:46</TIME>
        <TEMP>-60.1</TEMP>
        <DEWP>-58.1</DEWP>
        <STP>1034.5</STP>
        <SLP>1007.6</SLP>
        <VISIB>123.7</VISIB>
        <WDSP>10.8</WDSP>
        <PRCP>11.28</PRCP>
        <SNDP>11.1</SNDP>
        <FRSHTT>010101</FRSHTT>
        <CLDC>87.4</CLDC>
        <WNDDIR>342</WNDDIR>
    </MEASUREMENT>
</WEATHERDATA>

The goal of this project is to take this XML and parse it, that includes adding any missing data and correcting values that don't make sense. Then were supposed to take this data and save it on a server in as small a format as possible. We only have 20GB to work with.

I ended up making a little java application that I think works pretty well but i'm new to java so I really want to know if there is anything I can improve especially regarding performance.

This is the projects main file. Here I start the application and all of its thread pools. I also initialize a bunch of weather Station obj to use later.

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class WeatherService {
    private static final int SERVER_PORT = 7789;
    private static final boolean DEBUG = true;

    private WeatherService(ServerSocket serverSocket) {
        BlockingQueue<ArrayList<String>> parseQueue = new ArrayBlockingQueue<>(100000, true);
        BlockingQueue<Measurement> storeQueue = new ArrayBlockingQueue<>(100000, true);

        AtomicInteger parseCounter = new AtomicInteger(0);
        AtomicInteger storeCounter = new AtomicInteger(0);

        //TODO: performance can be improved
        Station[] weatherStations = new Station[1000000];
        initializeWeahterStations(weatherStations);

        // Create receiver thread pool
        ExecutorService receiverThreadPool = Executors.newCachedThreadPool();

        //Create and start parser thread pool
        ExecutorService parserThreadPool = Executors.newFixedThreadPool(2);
        parserThreadPool.submit(new ParserThread(parseQueue, parseCounter, storeQueue, weatherStations));

        //Create and start transmission thread pool
        ExecutorService transmissionThreadPool = Executors.newFixedThreadPool(2);
        transmissionThreadPool.submit(new TransmissionThread(storeQueue, storeCounter));

        //Create and start a performance control thread pool
        if (DEBUG){
            ScheduledExecutorService performanceControlPool = Executors.newScheduledThreadPool(1);
            performanceControlPool.scheduleAtFixedRate(new PerformanceControl(parseQueue, storeQueue, parseCounter, storeCounter), 0, 10, TimeUnit.SECONDS);
        }

        //Start receiver thread
        while (true) {
            try {
                Socket socket = serverSocket.accept();
                receiverThreadPool.submit(new ReceiverThread(socket, parseQueue));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void initializeWeahterStations(Station[] weatherStations) {
        for (int i = 0; i < 1000000; i++) {
            weatherStations[i] = new Station();
        }
    }

    public static void main(String[] args) {
        try {
            new WeatherService(new ServerSocket(SERVER_PORT));
        } catch (IOException e) {
            System.out.println("Error opening socket");
        }
    }
}

Station and measurement classes

import java.util.ArrayList;

public class Station {
    private ArrayList<Measurement> measurements = new ArrayList<>();

    Station() {
    }

    public void addNewMeasurement(Measurement measurement) {
        if (this.measurements.size() >= 30)
            this.measurements.remove(0);
        this.measurements.add(measurement);
    }

    public String getLatestFrshtt() {
        if (measurements.size() > 0)
            return measurements.get(0).frshtt;
        return "000000";
    }

    public double getExtrapolatedValue(int pos) {
        double total = 0;
        for (Measurement measurement : measurements) {
            total += measurement.measurementValues[pos];
        }
        return total / this.measurements.size();
    }

    public double getExtrapolatedTemp(double temp) {
        double total = 0;
        for (Measurement measurement : measurements) {
            total += measurement.measurementValues[0];
        }

        if (temp >= total * 0.8 || temp <= total * 1.2)
            return temp;
        else if (temp <= total * 0.8)
            return temp + (total * 0.8);
        else if (temp >= total * 1.2)
            return temp + (total * 1.2);

        return total / this.measurements.size();
    }
}

public class Measurement {
    public int[] measurementValues = new int[10];
    // [0] temp,    temperatuur in graden Celsius, geldige waardes van -9999.9 t/m 9999.9 met 1 decimaal
    // [1] dewp,    dauwpunt in graden Celsius, geldige waardes van -9999.9 t/m 9999.9 met 1 decimaal
    // [2] stp,     luchtdruk op stationsniveau in millibar, geldige waardes van 0.0 t/m 9999.9 met 1 decimaal
    // [3] slp,     luchtdruk op zeeniveau in millibar, geldige waardes van 0.0 t/m 9999.9 met 1 decimaal
    // [4] visib,   zichtbaarheid in kilometers, geldige waardes van 0.0 t/m 999.9 met 1 decimaal
    // [5] wdsp,    windsnelheid in kilometers per uur, geldige waardes van 0.0 t/m 999.9 met 1 decimaal
    // [6] prcp,    neerslag in centimeters, geldige waardes van 0.00 t/m 99999. met 2 decimalen
    // [7] sndp,    gevallen sneeuw in centimeters, geldige waardes van -9999.9 t/m 9999.9 met 1 decimaal
    // [8] cldc,    bewolking in procenten, geldige waardes van 0.0 t/m 99.9 met 1 decimaal
    // [9] wnddir,  windrichting in graden, geldige waardes van 0 t/m 359 alleen gehele getallen

    int stn;
    String frshtt;
    String date;    // Datum van versturen van deze gegevens, formaat: yyyy-mm-dd
    String time;    // Tijd van versturen van deze gegevens, formaat: hh:mm:ss

    public Measurement() {
    }
}

Each receiver thread connects to one weather cluster and receives 10 measurements per second in a single XML file. It only grabs the relevant lines and adds these to an array. The arrays get added to the parserQueue for later processing.

import java.io.*;
import java.net.Socket;
import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;

public class ReceiverThread implements Runnable {
    private final BlockingQueue<ArrayList<String>> outgoingQueue;
    private BufferedReader bufferedReader = null;
    private Socket socket;

    ReceiverThread(Socket socket, BlockingQueue<ArrayList<String>> outgoingQueue) {
        this.outgoingQueue = outgoingQueue;
        this.socket = socket;

        try {
            this.bufferedReader = new BufferedReader(new InputStreamReader(new BufferedInputStream(socket.getInputStream(), 256), "UTF-8"), 256);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void run() {
        try {
            this.receiverService();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void receiverService() throws IOException {
        try {
            ArrayList<String> measurement = new ArrayList<>();

            while (true) {
                String inputLine = bufferedReader.readLine();

                if (inputLine == null)
                    break;

                if (!inputLine.contains("?xml") && !inputLine.contains("<WEATHERDATA>") && !inputLine.contains("</WEATHERDATA>") && !inputLine.contains("<MEASUREMENT>") && !inputLine.contains("</MEASUREMENT>"))
                    measurement.add(inputLine.trim());

                if (inputLine.contains("</MEASUREMENT>")) {
                    outgoingQueue.add(measurement);
                    measurement = new ArrayList<>();
                }
            }
        } finally {
            bufferedReader.close();
            socket.close();
        }
    }
}

The two parser threads grab the arrays from the queue and start stripping the unnecessary characters. The values taken from the arrays get turned into a measurement object and added to the transmission queue.

In this thread the values also get checked if they are valid. Up to 30 measurements get added to each station to be used for extrapolation.

import java.util.ArrayList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class ParserThread implements Runnable {
    private final BlockingQueue<ArrayList<String>> incomingQueue;
    private final BlockingQueue<Measurement> outgoingQueue;
    private final AtomicInteger counter;
    private final Station[] weatherStations;

    ParserThread(BlockingQueue<ArrayList<String>> incomingQueue, AtomicInteger counter, BlockingQueue<Measurement> outgoingQueue, Station[] weatherStations) {
        this.incomingQueue = incomingQueue;
        this.outgoingQueue = outgoingQueue;
        this.counter = counter;
        this.weatherStations = weatherStations;
    }

    public void run() {
        while (true) {
            this.parseData();
        }
    }

    private void parseData() {
        try {
            if (incomingQueue.size() > 100) {
                ArrayList<ArrayList<String>> parseList = new ArrayList<>();
                this.incomingQueue.drainTo(parseList, 25000);

                for (ArrayList<String> items : parseList) {
                    Measurement measurement = new Measurement();
                    ItemsToMeasurement(items, measurement);

                    weatherStations[measurement.stn].addNewMeasurement(measurement);
                    this.outgoingQueue.add(measurement);
                    this.counter.getAndIncrement();
                }
            } else {
                Thread.sleep(2000);
            }

        } catch (Exception | Error e) {
            e.printStackTrace();
        }
    }

    //TODO: Make this shorter/ more efficient
    private void ItemsToMeasurement(ArrayList<String> items, Measurement measurement) {
        measurement.stn = Integer.parseInt(items.get(0).substring(5, items.get(0).length() - 6));
        measurement.date = items.get(1).substring(6, items.get(1).length() - 7);
        measurement.time = items.get(2).substring(6, items.get(2).length() - 7);
        measurement.frshtt = validateFrshtt(items.get(11).substring(8, items.get(11).length() - 9), measurement);

        measurement.measurementValues[0] = validateLine(items.get(3).substring(6, items.get(3).length() - 7), measurement, 0);
        measurement.measurementValues[1] = validateLine(items.get(4).substring(6, items.get(4).length() - 7), measurement, 1);
        measurement.measurementValues[2] = validateLine(items.get(5).substring(5, items.get(5).length() - 6), measurement, 2);
        measurement.measurementValues[3] = validateLine(items.get(6).substring(5, items.get(6).length() - 6), measurement, 3);
        measurement.measurementValues[4] = validateLine(items.get(7).substring(7, items.get(7).length() - 8), measurement, 4);
        measurement.measurementValues[5] = validateLine(items.get(8).substring(6, items.get(8).length() - 7), measurement, 5);
        measurement.measurementValues[6] = validateLine(items.get(9).substring(6, items.get(9).length() - 7), measurement, 6);
        measurement.measurementValues[7] = validateLine(items.get(10).substring(6, items.get(10).length() - 7), measurement, 7);
        measurement.measurementValues[8] = validateLine(items.get(12).substring(6, items.get(12).length() - 7), measurement, 8);
        measurement.measurementValues[9] = validateLine(items.get(13).substring(8, items.get(13).length() - 9), measurement, 9);
    }

    private String validateFrshtt(String line, Measurement measurement) {
        if (line.isEmpty())
            return weatherStations[measurement.stn].getLatestFrshtt();
        return line;
    }

    private int validateLine(String line, Measurement measurement, int pos) {
        if (line.isEmpty())
            return (int) (weatherStations[measurement.stn].getExtrapolatedValue(pos) * 100);

        if (pos == 0)
            return (int) (weatherStations[measurement.stn].getExtrapolatedTemp((Double.parseDouble(line)) * 100));

        return (int) (Double.parseDouble(line) * 100);
    }
}

Last but not least the measurements get written to a file. This part isn't quite done yet as I have yet to do anything to reduce the size of the data but ill get to that later. For now it writes integers to a file. It creates a folder for each day and in that folder a folder for each hour. In that folder a file gets created for each minute of measurements.

import java.io.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class TransmissionThread implements Runnable {
    private final BlockingQueue<Measurement> incomingQueue;
    private final AtomicInteger counter;

    TransmissionThread(BlockingQueue<Measurement> incomingQueue, AtomicInteger counter) {
        this.incomingQueue = incomingQueue;
        this.counter = counter;
    }

    public void run() {
        while (true) {
            this.transmitData();
        }
    }

    private void transmitData() {
        try {
            if (incomingQueue.size() > 100) {
                ArrayList<Measurement> transmitList = new ArrayList<>();
                this.incomingQueue.drainTo(transmitList, 25000);

                File directory = new File("C:\\Users\\djurr\\Desktop\\New folder\\out\\" + transmitList.get(0).date + "\\H" + transmitList.get(0).time.substring(0, 2));
                if (!directory.exists())
                    directory.mkdir();

                BufferedWriter writer = new BufferedWriter(
                        new FileWriter("C:\\Users\\djurr\\Desktop\\New folder\\out\\" + transmitList.get(0).date + "\\H" + transmitList.get(0).time.substring(0, 2) + "\\M"+ transmitList.get(0).time.substring(3, 5) +".txt", true)
                );

                for (Measurement measurement : transmitList) {
                    //TODO: bitify data
                    dataToFile(measurement, writer);
                    this.counter.getAndIncrement();
                }
                writer.close();

            } else {
                Thread.sleep(2000);
            }

        } catch (Exception | Error e) {
            e.printStackTrace();
        }
    }

    private void dataToFile(Measurement measurement, BufferedWriter writer) {
        try {
            writer.newLine();
            writer.write(measurement.stn + "," + measurement.time + "," + measurement.frshtt + "," + Arrays.toString(measurement.measurementValues));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

So yea if you see anything that I can do to improve this code, especially performance wise, id love to hear it. And sorry for the long post :P

Answer 1

Some hints about Java standard coding practices:

Declaring from Interface type instead of Class

In your code are present declarations like this below:

 private ArrayList<Measurement> measurements = new ArrayList<>();

Better use instead:

private List<Measurement> measurements = new ArrayList<>();

It allows flexibility if you do not want to implement an ArrayList but maybe another type of list without correcting every declaration in your code.

I have seen in your code you have the following if-else block:

if (temp >= total * 0.8 || temp <= total * 1.2)
       return temp;
else if (temp <= total * 0.8)
            return temp + (total * 0.8);
     else if (temp >= total * 1.2)
            return temp + (total * 1.2);
return total / this.measurements.size();

If you have an if branch ending with a return , you have no need to use the else for the other branch; refactoring these lines and with the help of local variables you can obtain:

double a = total * 0.8;
double b = total * 1.2;
if (temp >= a || temp <= b) { return temp; }
if (temp <= a) { return temp + a; }
if (temp >= b) { return temp + b; }
return total / this.measurements.size();

You have no need of use this in instance methods like your code below:

public class Station {
    private ArrayList<Measurement> measurements = new ArrayList<>();
    Station() {}
    public void addNewMeasurement(Measurement measurement) {
        if (this.measurements.size() >= 30)
            this.measurements.remove(0);
        this.measurements.add(measurement);
    }
}

You can rewrite your code initializing measurements inside the constructor and eliminate the this keyword from your methods:

public class Station {
    private List<Measurement> measurements;

    public Station() {
        this.measurements = new ArrayList<Measurement>();
    }

    public void addNewMeasurement(Measurement measurement) {
        if (measurements.size() >= 30) {
            measurements.remove(0);
        }
        measurements.add(measurement);
    }
}

About the reading of xml files, you are not using one of the xml api Java offers, I'm including an example of how you can read the file you posted using the Java dom package and extract for example the DATE and TIME tags:

File file = new File("measurement.xml");
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
DocumentBuilder builder = factory.newDocumentBuilder();
Document doc = builder.parse(file);
Node node = doc.getElementsByTagName("DATE").item(0);
String date = node.getTextContent();
node = doc.getElementsByTagName("TIME").item(0);
String time = node.getTextContent();
LocalDateTime dateTime = LocalDateTime.of(LocalDate.parse(date), LocalTime.parse(time));
System.out.println(dateTime.toLocalTime());
System.out.println(dateTime.toLocalDate());

You can start from this example to extract other tags inside the xml file, instead of using two distinct fields for memorizing date and time in your class Measurement you have only to store the LocalDateTime object.