package au.gov.amsa.navigation.ais;

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

import rx.Observable;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import au.gov.amsa.navigation.DriftingDetector;
import au.gov.amsa.navigation.VesselPosition;
import au.gov.amsa.navigation.VesselPositions;
import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.util.Files;

import com.github.davidmoten.rx.slf4j.Logging;

public class BinaryFixesDriftDetectorMain {

	public static void main(String[] args) {
		// perform a speed test for loading BinaryFixes from disk

		// -downsample-5-mins
		List<File> files = Files.find(new File("/media/an/binary-fixes/2014-year"),
		        Pattern.compile(".*\\.track"));
		int count = Observable
		// list files
		        .from(files)
		        // share the load between processors
		        .buffer(Math.max(1, files.size() / Runtime.getRuntime().availableProcessors()))
		        // count each file asynchronously
		        .flatMap(new Func1<List<File>, Observable<VesselPosition>>() {
			        @Override
			        public Observable<VesselPosition> call(List<File> list) {
				        return Observable.from(list)
				                .concatMap(new Func1<File, Observable<VesselPosition>>() {

					                @Override
					                public Observable<VesselPosition> call(File file) {
						                return new DriftingDetector().getCandidates(
						                        BinaryFixes.from(file).map(
						                                VesselPositions.TO_VESSEL_POSITION))
						                // log
						                        .lift(Logging.<VesselPosition> logger().showValue()
						                                .log());

					                }
				                })
				                // schedule
				                .subscribeOn(Schedulers.computation());
			        }
		        }).count().toBlocking().single();
		System.out.println("drift detections = " + count);
	}
}

@@ -695,18 +695,17 @@
 ixes
-/2014-year
+-all/2012
 %22),%0A
@@ -1430,16 +1430,47 @@
 tector()
+%0A%09%09%09%09%09%09                        
 .getCand
@@ -1507,16 +1507,24 @@
         
+    
+    
 BinaryFi
@@ -1526,16 +1526,63 @@
 aryFixes
+%0A%09%09%09%09%09%09                                        
 .from(fi
@@ -1588,13 +1588,8 @@
 ile)
-.map(
 %0A%09%09%09
@@ -1623,16 +1623,29 @@
         
+        .map(
 VesselPo
@@ -1671,24 +1671,1020 @@
 OSITION)
-)
 %0A%09%09%09%09%09%09
+                                        .filter(new Func1%3CVesselPosition, Boolean%3E() %7B%0A%0A%09%09%09%09%09%09%09                                        @Override%0A%09%09%09%09%09%09%09                                        public Boolean call(%0A%09%09%09%09%09%09%09                                                VesselPosition p) %7B%0A%09%09%09%09%09%09%09%09                                        return p.cogDegrees()%0A%09%09%09%09%09%09%09%09                                                .isPresent()%0A%09%09%09%09%09%09%09%09                                                && p.headingDegrees()%0A%09%09%09%09%09%09%09%09                                                        .isPresent()%0A%09%09%09%09%09%09%09%09                                                && p.speedMetresPerSecond()%0A%09%09%09%09%09%09%09%09                                                        .isPresent();%0A%09%09%09%09%09%09%09                                        %7D%0A%0A%09%09%09%09%09%09                                        %7D)%0A%09%09%09%09%09%09                                        .lift(Logging%0A%09%09%09%09%09%09                                                .%3CVesselPosition%3E log()))%0A%09%09%09%09%09%09        
         

add comment