package au.gov.amsa.geo.adhoc;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.PrintStream;
import java.util.List;
import java.util.regex.Pattern;

import rx.Observable;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import au.gov.amsa.ais.AisMessage;
import au.gov.amsa.ais.Timestamped;
import au.gov.amsa.ais.message.AisShipStatic;
import au.gov.amsa.ais.message.AisShipStaticA;
import au.gov.amsa.ais.rx.Streams;
import au.gov.amsa.ais.rx.Streams.TimestampedAndLine;
import au.gov.amsa.util.Files;

import com.github.davidmoten.rx.slf4j.Logging;
import com.google.common.base.Optional;

public class ShipStaticDataMain {

    public static void main(String[] args) throws FileNotFoundException {
        PrintStream out = new PrintStream("target/out.txt");
        // String filename = "/media/an/nmea/2014/NMEA_ITU_20140815.gz";
        List<File> files = Files.find(new File("/media/an/nmea/2014/"),
                Pattern.compile("NMEA_ITU.*.gz"));

        Observable
                .from(files)
                .buffer(Math.max(1, files.size() / Runtime.getRuntime().availableProcessors()))
                .flatMap(
                        list -> Observable.from(list)
                                .lift(Logging.<File> logger().showValue().showMemory().log())
                                .flatMap(file -> Streams.extract(Streams.nmeaFromGzip(file))
                                //
                                // .lift(Logging
                                // .<TimestampedAndLine<AisMessage>>
                                // logger()
                                // .showCount().every(1000000).log())
                                //
                                        .flatMap(aisShipStaticOnly)
                                        //
                                        .map(m -> m.getMessage().get().message())
                                        //
                                        .filter(m -> m instanceof AisShipStaticA)
                                        //
                                        .cast(AisShipStaticA.class)
                                        //
                                        .distinct(m -> m.getMmsi()))
                                //
                                .subscribeOn(Schedulers.computation()))
                //
                .toMap(m -> m.getMmsi(), m -> m)
                //
                .doOnNext(
                        map -> {
                            System.out.println("writing map to target/out.txt");
                            map.values()
                                    .stream()
                                    .forEach(
                                            m -> {
                                                out.format(
                                                        "%s,%s,%s,%s,%s,%s,%s,%s\n",
                                                        m.getMmsi(),
                                                        m.getImo().or(-1),
                                                        "A",
                                                        m.getShipType(),
                                                        (float) m
                                                                .getMaximumPresentStaticDraughtMetres(),
                                                        m.getDimensionA().or(-1), m.getDimensionB()
                                                                .or(-1), m.getDimensionC().or(-1),
                                                        m.getDimensionD().or(-1));
                                            });
                            ;
                        }).toBlocking().single();
        out.close();

    }

    private static Func1<TimestampedAndLine<AisMessage>, Observable<TimestampedAndLine<AisShipStatic>>> aisShipStaticOnly = m -> {
        Optional<Timestamped<AisMessage>> message = m.getMessage();
        if (message.isPresent() && message.get().message() instanceof AisShipStatic) {
            @SuppressWarnings("unchecked")
            TimestampedAndLine<AisShipStatic> m2 = (TimestampedAndLine<AisShipStatic>) (TimestampedAndLine<?>) m;
            return Observable.just(m2);
        } else
            return Observable.empty();
    };
}

@@ -1132,16 +1132,20 @@
 ssors())
+ - 1
 )%0A      
@@ -1341,26 +1341,28 @@
            .
-fl
+conc
 atMap(file -
@@ -2322,88 +2322,16 @@
    .
-subscribeOn(Schedulers.computation()))%0A                //%0A                .toMap
+distinct
 (m -
@@ -2347,16 +2347,8 @@
 si()
-, m -%3E m
 )%0A  
@@ -2353,35 +2353,32 @@
                 
-//%0A
                 
@@ -2381,18 +2381,10 @@
     
-.doOnNext(
+//
 %0A   
@@ -2408,137 +2408,54 @@
     
-map -%3E %7B%0A                            System.out.println(%22writing map to target/out.txt%22);%0A                            map.values(
+        .subscribeOn(Schedulers.computation())
 )%0A  
@@ -2460,36 +2460,35 @@
                 
-    
+//%0A
                 
@@ -2492,15 +2492,33 @@
    .
-stream(
+distinct(m -%3E m.getMmsi()
 )%0A  
@@ -2523,32 +2523,35 @@
                 
+//%0A
                 
@@ -2554,42 +2554,19 @@
     
-    .forEach(%0A                    
+.doOnNext(%0A
     
@@ -2624,249 +2624,60 @@
     
-                    out.format(%0A                                                        %22%25s,%25s,%25s,%25s,%25s,%25s,%25s,%25s%5Cn%22,%0A                                                        m.getMmsi(),%0A                                                       
+out.format(%22%25s,%25s,%25s,%25s,%25s,%25s,%25s,%25s%5Cn%22, m.getMmsi(),
  m.g
@@ -2732,192 +2732,40 @@
     
-          
-          %22A%22,%0A                                                        m.getShipType(),%0A                                                        (float) m%0A                    
+%22A%22, m.getShipType(), (float) m%0A
     
@@ -2844,29 +2844,19 @@
 etres(),
-%0A            
+ m%0A
         
@@ -2883,33 +2883,32 @@
                 
-m
 .getDimensionA()
@@ -2933,37 +2933,27 @@
 nsionB()
-%0A                    
+.or(-1), m%0A
         
@@ -2988,26 +2988,16 @@
         
-.or(-1), m
 .getDime
@@ -3016,64 +3016,8 @@
 -1),
-%0A                                                       
  m.g
@@ -3068,33 +3068,26 @@
         
-                %7D
+out.flush(
 );%0A     
@@ -3109,13 +3109,32 @@
     
-    ;
+%7D)%0A                // go
 %0A   
@@ -3146,25 +3146,23 @@
         
-        %7D
+.count(
 ).toBloc
@@ -3199,17 +3199,16 @@
 lose();%0A
-%0A
     %7D%0A%0A 

improve static ship data extract