<commit_before>package au.gov.amsa.navigation.ais;

import java.io.File;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rx.Observable;
import rx.Scheduler;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;
import au.gov.amsa.navigation.DriftCandidate;
import au.gov.amsa.navigation.DriftingDetectorFix;
import au.gov.amsa.navigation.VesselPosition;
import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.risky.format.Downsample;
import au.gov.amsa.risky.format.Fix;
import au.gov.amsa.util.Files;
import au.gov.amsa.util.identity.MmsiValidator2;

public class BinaryFixesDriftDetectorMain {

	private static final Logger log = LoggerFactory.getLogger(BinaryFixesDriftDetectorMain.class);

	public static void main(String[] args) {

		VesselPosition.validate = false;
		Fix.validate = false;
		List<File> files = Files.find(new File("/media/an/binary-fixes-2014"),
		        Pattern.compile(".*\\.track"));
		log.info("files=" + files.size());
		final AtomicLong num = new AtomicLong();
		int count = Observable
		// list files
		        .from(files).filter(onlyValidMmsis())
		        // share the load between processors
		        .buffer(Math.max(1, files.size() / Runtime.getRuntime().availableProcessors() - 1))
		        // search each list of files for drift detections
		        .flatMap(detectDrift(num, Schedulers.computation()))
		        // count
		        .reduce(0, BinaryFixesDriftDetectorMain.<Integer> add()).toBlocking().single();
		log.info("drift detections = " + count);
	}

	private static Func1<? super File, Boolean> onlyValidMmsis() {
		return new Func1<File, Boolean>() {

			@Override
			public Boolean call(File file) {
				return MmsiValidator2.INSTANCE.isValid(Long.parseLong(file.getName().substring(0,
				        file.getName().indexOf('.'))));
			}
		};
	}

	private static <T extends Number> Func2<T, T, T> add() {
		return new Func2<T, T, T>() {
			@SuppressWarnings("unchecked")
			@Override
			public T call(T a, T b) {
				if (a instanceof Integer)
					return (T) (Number) (a.intValue() + b.intValue());
				else if (a instanceof Long)
					return (T) (Number) (a.longValue() + b.longValue());
				else if (a instanceof Double)
					return (T) (Number) (a.doubleValue() + b.doubleValue());
				else if (a instanceof Float)
					return (T) (Number) (a.floatValue() + b.floatValue());
				else if (a instanceof Byte)
					return (T) (Number) (a.byteValue() + b.byteValue());
				else if (a instanceof Short)
					return (T) (Number) (a.shortValue() + b.shortValue());
				else
					throw new RuntimeException("not implemented");
			}
		};
	}

	private static Func1<List<File>, Observable<Integer>> detectDrift(final AtomicLong num,
	        final Scheduler scheduler) {
		return new Func1<List<File>, Observable<Integer>>() {
			@Override
			public Observable<Integer> call(List<File> list) {
				return Observable.from(list).concatMap(new Func1<File, Observable<Integer>>() {
					@Override
					public Observable<Integer> call(File file) {
						return BinaryFixes.from(file)
						// log count
						        .doOnNext(logCount(num))
						        // detect drift
						        .compose(DriftingDetectorFix.detectDrift())
						        // downsample to min 5 minutes between reports
						        .compose(Downsample.minTimeStep(5, TimeUnit.MINUTES))
						        // cast
						        .cast(DriftCandidate.class)
						        // onNext
						        .doOnNext(onEach())
						        // count
						        .count();
					}

				})
				// schedule
				        .subscribeOn(scheduler);
			}
		};
	}

	private static Action1<DriftCandidate> onEach() {
		return new Action1<DriftCandidate>() {

			@Override
			public void call(DriftCandidate c) {
				log.info(c.toString());
			}
		};
	}

	private static Action1<Fix> logCount(final AtomicLong num) {
		return new Action1<Fix>() {
			@Override
			public void call(Fix p) {
				long n = num.incrementAndGet();
				if (n % 1000000 == 0)
					log.info((n / 1000000.0) + "m");
			}
		};
	}

}<commit_msg>drift detections write to a file<commit_after>package au.gov.amsa.navigation.ais;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rx.Observable;
import rx.Scheduler;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;
import au.gov.amsa.navigation.DriftCandidate;
import au.gov.amsa.navigation.DriftingDetectorFix;
import au.gov.amsa.navigation.VesselPosition;
import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.risky.format.Downsample;
import au.gov.amsa.risky.format.Fix;
import au.gov.amsa.util.Files;
import au.gov.amsa.util.identity.MmsiValidator2;

public class BinaryFixesDriftDetectorMain {

	private static final String DRIFT_CANDIDATES_TXT = "target/drift-candidates.txt";
	private static final Logger log = LoggerFactory.getLogger(BinaryFixesDriftDetectorMain.class);

	public static void main(String[] args) {
		new File(DRIFT_CANDIDATES_TXT).delete();
		VesselPosition.validate = false;
		Fix.validate = false;
		List<File> files = Files.find(new File("/media/an/binary-fixes-2014"),
		        Pattern.compile(".*\\.track"));
		log.info("files=" + files.size());
		final AtomicLong num = new AtomicLong();
		int count = Observable
		// list files
		        .from(files).filter(onlyValidMmsis())
		        // share the load between processors
		        .buffer(Math.max(1, files.size() / Runtime.getRuntime().availableProcessors() - 1))
		        // search each list of files for drift detections
		        .flatMap(detectDrift(num, Schedulers.computation()))
		        // count
		        .reduce(0, BinaryFixesDriftDetectorMain.<Integer> add()).toBlocking().single();
		log.info("drift detections = " + count);
	}

	private static Func1<? super File, Boolean> onlyValidMmsis() {
		return new Func1<File, Boolean>() {

			@Override
			public Boolean call(File file) {
				return MmsiValidator2.INSTANCE.isValid(Long.parseLong(file.getName().substring(0,
				        file.getName().indexOf('.'))));
			}
		};
	}

	private static <T extends Number> Func2<T, T, T> add() {
		return new Func2<T, T, T>() {
			@SuppressWarnings("unchecked")
			@Override
			public T call(T a, T b) {
				if (a instanceof Integer)
					return (T) (Number) (a.intValue() + b.intValue());
				else if (a instanceof Long)
					return (T) (Number) (a.longValue() + b.longValue());
				else if (a instanceof Double)
					return (T) (Number) (a.doubleValue() + b.doubleValue());
				else if (a instanceof Float)
					return (T) (Number) (a.floatValue() + b.floatValue());
				else if (a instanceof Byte)
					return (T) (Number) (a.byteValue() + b.byteValue());
				else if (a instanceof Short)
					return (T) (Number) (a.shortValue() + b.shortValue());
				else
					throw new RuntimeException("not implemented");
			}
		};
	}

	private static Func1<List<File>, Observable<Integer>> detectDrift(final AtomicLong num,
	        final Scheduler scheduler) {
		return new Func1<List<File>, Observable<Integer>>() {
			@Override
			public Observable<Integer> call(List<File> list) {
				return Observable.from(list).concatMap(new Func1<File, Observable<Integer>>() {
					@Override
					public Observable<Integer> call(File file) {
						return BinaryFixes.from(file)
						// log count
						        .doOnNext(logCount(num))
						        // detect drift
						        .compose(DriftingDetectorFix.detectDrift())
						        // downsample to min 5 minutes between reports
						        .compose(
						                Downsample
						                        .<DriftCandidate> minTimeStep(5, TimeUnit.MINUTES))
						        // onNext
						        .doOnNext(ON_NEXT)
						        // count
						        .count();
					}

				})
				// schedule
				        .subscribeOn(scheduler);
			}
		};
	}

	private static final Action1<DriftCandidate> ON_NEXT = new Action1<DriftCandidate>() {
		final char COMMA = ',';
		final Charset UTF8 = Charset.forName("UTF-8");

		@Override
		public synchronized void call(DriftCandidate c) {
			try {
				FileOutputStream os = new FileOutputStream(DRIFT_CANDIDATES_TXT, true);
				Fix f = c.fix();
				StringBuilder s = new StringBuilder();
				s.append(f.getMmsi());
				s.append(COMMA);
				s.append(f.getLat());
				s.append(COMMA);
				s.append(f.getLon());
				s.append(COMMA);
				s.append(f.getTime());
				s.append(COMMA);
				s.append(f.getAisClass());
				s.append(COMMA);
				s.append(f.getCourseOverGroundDegrees().get());
				s.append(COMMA);
				s.append(f.getHeadingDegrees().get());
				s.append(COMMA);
				s.append(f.getSpeedOverGroundKnots().get());
				s.append(COMMA);
				s.append(f.getNavigationalStatus().isPresent() ? f.getNavigationalStatus().get()
				        .toString() : "");
				s.append(COMMA);
				s.append(c.driftingSince());
				s.append(COMMA);
				s.append('\n');
				os.write(s.toString().getBytes(UTF8));
				os.close();
			} catch (IOException e) {
				throw new RuntimeException(e);
			}
		}
	};

	private static Action1<Fix> logCount(final AtomicLong num) {
		return new Action1<Fix>() {
			@Override
			public void call(Fix p) {
				long n = num.incrementAndGet();
				if (n % 1000000 == 0)
					log.info((n / 1000000.0) + "m");
			}
		};
	}

}
{'fraction_non_alphanumeric': 0.13856168629882207, 'fraction_numerical': 0.009402769167183302, 'mean_word_length': 3.103009749894023, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 48, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7155497', 'n_tokens_mistral': 3756, 'n_tokens_neox': 3108, 'n_words': 673}