@@ -0,0 +1,2760 @@
+package au.gov.amsa.geo.adhoc;%0A%0Aimport java.io.BufferedWriter;%0Aimport java.io.File;%0Aimport java.io.FileOutputStream;%0Aimport java.io.IOException;%0Aimport java.io.OutputStreamWriter;%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0Aimport java.util.concurrent.atomic.AtomicInteger;%0Aimport java.util.regex.Pattern;%0A%0Aimport au.gov.amsa.risky.format.BinaryFixes;%0Aimport au.gov.amsa.risky.format.Fix;%0Aimport au.gov.amsa.util.Files;%0Aimport au.gov.amsa.util.identity.MmsiValidator2;%0Aimport rx.Observable;%0Aimport rx.Scheduler;%0Aimport rx.observables.GroupedObservable;%0Aimport rx.schedulers.Schedulers;%0A%0Apublic class VesselsInCoverageMain %7B%0A%0A    public static void main(String%5B%5D args) throws IOException %7B%0A        long t = System.currentTimeMillis();%0A        File out = new File(%22target/mmsi.txt%22);%0A        out.delete();%0A        try (BufferedWriter writer = new BufferedWriter(%0A                new OutputStreamWriter(new FileOutputStream(out)))) %7B%0A            Pattern pattern = Pattern.compile(%22.*%5C%5C.track%22);%0A            List%3CFile%3E list = new ArrayList%3CFile%3E();%0A            list.addAll(Files.find(new File(%22/media/an/binary-fixes-5-minute/2014%22), pattern));%0A            list.addAll(Files.find(new File(%22/media/an/binary-fixes-5-minute/2015%22), pattern));%0A            list.addAll(Files.find(new File(%22/media/an/binary-fixes-5-minute/2016%22), pattern));%0A            AtomicInteger count = new AtomicInteger();%0A%0A            Observable.from(list) //%0A                    .groupBy(f -%3E count.getAndIncrement()%0A                            %25 Runtime.getRuntime().availableProcessors()) //%0A                    .flatMap(files -%3E toFixes(files, Schedulers.computation())) //%0A                    .distinct(fix -%3E fix.mmsi() + fix.aisClass().name()) //%0A                    .sorted((a, b) -%3E Integer.compare(a.mmsi(), b.mmsi())) //%0A                    .filter(fix -%3E MmsiValidator2.INSTANCE.isValid((long) fix.mmsi())) //%0A                    .doOnNext(fix -%3E write(writer, fix)) //%0A                    .toBlocking() //%0A                    .subscribe();%0A        %7D%0A        System.out.println((System.currentTimeMillis() - t) + %22ms%22);%0A    %7D%0A%0A    private static void write(BufferedWriter b, Fix fix) %7B%0A        try %7B%0A            b.write(String.valueOf(fix.mmsi()));%0A            b.write('%5Ct');%0A            b.write(fix.aisClass().name());%0A            b.write('%5Cn');%0A        %7D catch (IOException e) %7B%0A            throw new RuntimeException(e);%0A        %7D%0A    %7D%0A%0A    private static Observable%3CFix%3E toFixes(GroupedObservable%3CInteger, File%3E files,%0A            Scheduler scheduler) %7B%0A        return files //%0A                .flatMap(f -%3E BinaryFixes.from(f) //%0A                        .distinct(fix -%3E fix.mmsi() + fix.aisClass().name()) //%0A                        .subscribeOn(scheduler));%0A    %7D%0A%0A%7D%0A

add ad-hoc query