<commit_before>package au.gov.amsa.geo.adhoc;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.TimeZone;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.zip.GZIPOutputStream;

import org.apache.log4j.Logger;

import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.risky.format.BinaryFixesFormat;
import au.gov.amsa.risky.format.Fix;
import rx.Observable;

public final class BinaryFixesWithMmsiGzCombinedSortMain {
    
    private static final Logger log = Logger.getLogger(BinaryFixesWithMmsiGzCombinedSortMain.class);

    public static void main(String[] args) throws InterruptedException {
        File tracks = new File("/home/dxm/AIS/tracks");
        File combinedSortedTracks = new File("/home/dxm/combinedSortedTracks");
        combinedSortedTracks.mkdir();

        AtomicReference<FileFixes> previous = new AtomicReference<>();
        AtomicInteger n = new AtomicInteger();
        Observable //
                .from(tracks.listFiles()) //
                .filter(x -> x.getName().endsWith(".track.gz") && x.getName().compareTo("2019-08-06") >= 0) //
                .map(x -> new FileFixes(x,
                        BinaryFixes.from(x, true, BinaryFixesFormat.WITH_MMSI).toList().toBlocking().first()))
                .doOnNext(ff -> {
                    FileFixes prev = previous.get();
                    if (prev != null) {
                        long removed = 0;
                        long added = 0;
                        Iterator<Fix> it = ff.fixes.iterator();
                        while (it.hasNext()) {
                            Fix fix = it.next();
                            if (fix.time() < ff.startTime) {
                                it.remove();
                                removed++;
                                if (fix.time() >= prev.startTime) {
                                    prev.fixes.add(fix);
                                    added++;
                                }
                            }
                        }
                        prev.fixes.sort((x, y) -> Long.compare(x.time(), y.time()));
                        File f = new File(combinedSortedTracks, prev.file.getName());
                        try (OutputStream out = new GZIPOutputStream(new FileOutputStream(f))) {
                            for (Fix fix : prev.fixes) {
                                BinaryFixes.write(fix, out, BinaryFixesFormat.WITH_MMSI);
                            }
                        } catch (IOException e) {
                            throw new UncheckedIOException(e);
                        }
                        log.info(n.incrementAndGet() + ": removed " + removed + ", added " + added + " to " + f);
                    }
                    previous.set(ff);
                }).subscribe();
    }

    private static final class FileFixes {
        final File file;
        final List<Fix> fixes;
        final long startTime;

        FileFixes(File file, List<Fix> fixes) {
            this.file = file;
            this.fixes = fixes;
            String datePart = file.getName().substring(0, 10);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-mm-dd");
            sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
            try {
                startTime = sdf.parse(datePart).getTime();
            } catch (ParseException e) {
                throw new RuntimeException(e);<commit_msg>use linkedlist to increase perf of removal<commit_after>package au.gov.amsa.geo.adhoc;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.TimeZone;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.GZIPOutputStream;

import org.apache.log4j.Logger;

import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.risky.format.BinaryFixesFormat;
import au.gov.amsa.risky.format.Fix;
import rx.Observable;

public final class BinaryFixesWithMmsiGzCombinedSortMain {
    
    private static final Logger log = Logger.getLogger(BinaryFixesWithMmsiGzCombinedSortMain.class);

    public static void main(String[] args) throws InterruptedException {
        File tracks = new File("/home/dxm/AIS/tracks");
        File combinedSortedTracks = new File("/home/dxm/combinedSortedTracks2");
        combinedSortedTracks.mkdir();

        AtomicReference<FileFixes> previous = new AtomicReference<>();
        AtomicInteger n = new AtomicInteger();
        Observable //
                .from(tracks.listFiles()) //
                .filter(x -> x.getName().endsWith(".track.gz") && x.getName().startsWith("2019-01-0")) //
                .sorted((x,y) -> x.getName().compareTo(y.getName())) //
                .map(x -> new FileFixes(x, BinaryFixes.from(x, true, BinaryFixesFormat.WITH_MMSI) //
                        // use a linked list because iterator.remove is efficient with that data structure
                        .reduce(new LinkedList<Fix>(), (list, fix) -> {
                            list.add(fix);
                            return list;
                        }) //
                        .toBlocking() //
                        .first()))
                .doOnNext(ff -> {
                    FileFixes prev = previous.get();
                    if (prev != null) {
                        long removed = 0;
                        long added = 0;
                        Iterator<Fix> it = ff.fixes.iterator();
                        while (it.hasNext()) {
                            Fix fix = it.next();
                            if (fix.time() < ff.startTime) {
                                it.remove();
                                removed++;
                                if (fix.time() >= prev.startTime) {
                                    prev.fixes.add(fix);
                                    added++;
                                }
                            }
                        }
                        prev.fixes.sort((x, y) -> Long.compare(x.time(), y.time()));
                        writeFixes(combinedSortedTracks, n, prev, removed, added);
                    }
                    previous.set(ff);
                }).subscribe();
        if (previous.get() != null) {
            writeFixes(combinedSortedTracks, n, previous.get(), 0, 0);
        }
    }

    private static void writeFixes(File combinedSortedTracks, AtomicInteger n, FileFixes prev, long removed,
            long added) {
        File f = new File(combinedSortedTracks, prev.file.getName());
        try (OutputStream out = new GZIPOutputStream(new FileOutputStream(f))) {
            for (Fix fix : prev.fixes) {
                BinaryFixes.write(fix, out, BinaryFixesFormat.WITH_MMSI);
            }
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
        log.info(n.incrementAndGet() + ": removed from next day=" + removed + ", added to this day=" + added + " to " + f);
    }

    private static final class FileFixes {
        final File file;
        final List<Fix> fixes;
        final long startTime;

        FileFixes(File file, List<Fix> fixes) {
            this.file = file;
            this.fixes = fixes;
            String datePart = file.getName().substring(0, 10);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-mm-dd");
            sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
            try {
                startTime = sdf.parse(datePart).getTime();
            } catch (ParseException e) {
                throw new RuntimeException(e);
{'fraction_non_alphanumeric': 0.11153281096963762, 'fraction_numerical': 0.0037952987267384918, 'mean_word_length': 1.621630295250321, 'pattern_counts': {'":': 2, '<': 16, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3528093', 'n_tokens_mistral': 2256, 'n_tokens_neox': 2133, 'n_words': 449}