Calculate distance to expert distribution layers

livingatlas/configs/la-pipelines-local-hdfs.yaml

@@ -66,3 +66,7 @@ sample-avro:
 index:
     inputPath: hdfs://localhost:8020/pipelines-data
     zkHost: localhost:9983
+outlier:
+    inputPath: hdfs://localhost:8020/pipelines-data
+    targetPath: hdfs://localhost:8020/pipelines-outlier
+

livingatlas/configs/la-pipelines.yaml

@@ -167,6 +167,15 @@ sampling:
   allDatasetsInputPath: '{fsPath}/pipelines-all-datasets'
   runner: SparkRunner
 
+# Calculate distance to the expert distribution layers
+# class:au.org.ala.pipelines.beam.DistributionOutlierPipeline
+outlier:
+  appName: Expert distribution outliers for {datasetId}
+  baseUrl: https://spatial.ala.org.au/ws/
+  targetPath: &outlierTargetPath '{fsPath}/pipelines-outlier'
+  allDatasetsInputPath: '{fsPath}/pipelines-all-datasets'
+  runner: SparkRunner
+
 # class: au.org.ala.pipelines.beam.ALAInterpretedToSensitivePipeline
 sensitive:
   appName: Sensitive Data Processing for {datasetId}
@@ -192,6 +201,7 @@ solr:
   allDatasetsInputPath: '{fsPath}/pipelines-all-datasets'
   jackKnifePath: "{fsPath}/pipelines-jackknife"
   clusteringPath: "{fsPath}/pipelines-clustering"
+  outlierPath: *outlierTargetPath
   solrCollection: biocache
   includeSampling: false
   includeJackKnife: false

livingatlas/pipelines/src/main/java/au/org/ala/distribution/DistributionLayer.java

@@ -0,0 +1,37 @@
+package au.org.ala.distribution;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.PropertyNamingStrategy;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Represent a field in the sampling service. A field is a property derived from a spatial layer.
+ */
+@Getter
+@Setter
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy.class)
+public class DistributionLayer {
+  private int gid;
+  private String data_resource_uid;
+  private String scientific;
+  private String pid;
+  private String type;
+  private String genus_name;
+  private String lsid;
+  private float area_km;
+  private String common_nam;
+  private int spcode;
+  private String genus_lsid;
+  private String bounding_box;
+  private String group_name;
+  private Boolean endemic;
+  private String image_url;
+  private String family_lsid;
+  private String specific_n;
+  private String wmsurl;
+  private String family;
+  private int geom_idx;
+}

livingatlas/pipelines/src/main/java/au/org/ala/distribution/DistributionService.java

@@ -0,0 +1,33 @@
+package au.org.ala.distribution;
+
+import java.util.List;
+import java.util.Map;
+import retrofit2.Call;
+import retrofit2.http.*;
+
+/** Simple client to the ALA sampling service. */
+public interface DistributionService {
+
+  /** Return expert distribution layers in the ALA spatial portal */
+  @GET("distribution/")
+  Call<List<DistributionLayer>> getLayers();
+
+  @GET("distribution/lsids/{id}")
+  Call<List<DistributionLayer>> getLayersByLsid(
+      @Path("id") String id, @Query("nowkt") String nowkt);
+
+  /**
+   * @param lsid
+   * @param points Map<uuid, <decimalLatitude,decimalLongitude>>
+   * @return
+   */
+  @POST("distribution/outliers/{id}")
+  Call<Map<String, Double>> outliers(
+      @Path("id") String lsid, @Body Map<String, Map<String, Double>> points);
+
+  //  @FormUrlEncoded
+  //  @POST("distribution/outliers/{id}")
+  //  Call<DistributionRequest> outliers(@Path(value="id",encoded = true) String lsid,
+  // @Field("pointsJson") Map<String, Map<String, Double>> points);
+
+}

livingatlas/pipelines/src/main/java/au/org/ala/distribution/DistributionServiceImpl.java

@@ -0,0 +1,91 @@
+package au.org.ala.distribution;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategy;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import retrofit2.Response;
+import retrofit2.Retrofit;
+import retrofit2.converter.jackson.JacksonConverterFactory;
+
+@Slf4j
+public class DistributionServiceImpl implements Serializable {
+
+  private Retrofit retrofit;
+  private DistributionService service;
+
+  private DistributionServiceImpl(String baseUrl) {
+    ObjectMapper om =
+        new ObjectMapper().setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
+    JacksonConverterFactory jcf = JacksonConverterFactory.create(om);
+    retrofit = new Retrofit.Builder().baseUrl(baseUrl).addConverterFactory(jcf).build();
+    service = retrofit.create(DistributionService.class);
+  }
+
+  public static DistributionServiceImpl init(String baseUrl) {
+    // set up sampling service
+    return new DistributionServiceImpl(baseUrl);
+  }
+
+  public List<DistributionLayer> getLayers() throws IOException, ExpertDistributionException {
+    // Response<List<DistributionLayer>> response =
+    // distributionService.getLayersByLsid("urn:lsid:biodiversity.org.au:afd.taxon:4f3a5260-4f39-4393-a644-4d05b1c45f92", "false").execute();
+    Response<List<au.org.ala.distribution.DistributionLayer>> response =
+        service.getLayers().execute();
+    int code = response.code();
+    if (code >= 200 && code < 300) {
+      List<au.org.ala.distribution.DistributionLayer> layers = response.body();
+      return layers;
+    } else {
+      errorHandler(code, response);
+      return null;
+    }
+  }
+
+  public List<DistributionLayer> findLayersByLsid(String lsid)
+      throws IOException, ExpertDistributionException {
+    Response<List<DistributionLayer>> response = service.getLayersByLsid(lsid, "false").execute();
+    int code = response.code();
+    if (code >= 200 && code < 300) {
+      List<au.org.ala.distribution.DistributionLayer> layers = response.body();
+      return layers;
+    } else {
+      errorHandler(code, response);
+      return null;
+    }
+  }
+
+  public Map<String, Double> outliers(String lsid, Map<String, Map<String, Double>> points)
+      throws IOException, ExpertDistributionException {
+    Response<Map<String, Double>> response = service.outliers(lsid, points).execute();
+    int code = response.code();
+    if (code >= 200 && code < 300) {
+      Map<String, Double> results = response.body();
+      return results;
+    } else {
+      errorHandler(code, response);
+      return null;
+    }
+  }
+
+  /**
+   * Todo Handle error information Need to sync with Spatial Service
+   *
+   * @param code
+   * @param response
+   * @throws IOException
+   * @throws ExpertDistributionException
+   */
+  private void errorHandler(int code, Response<?> response)
+      throws IOException, ExpertDistributionException {
+    String errorBody = response.errorBody().string();
+    if (code >= 400 && code < 500) {
+      throw new ExpertDistributionException(errorBody);
+    } else {
+      throw new RuntimeException(errorBody);
+    }
+  }
+}

livingatlas/pipelines/src/main/java/au/org/ala/distribution/ExpertDistributionException.java

@@ -0,0 +1,7 @@
+package au.org.ala.distribution;
+
+public class ExpertDistributionException extends Exception {
+  public ExpertDistributionException(String errorMessage) {
+    super(errorMessage);
+  }
+}

livingatlas/pipelines/src/main/java/au/org/ala/pipelines/beam/DistributionOutlierPipeline.java

@@ -0,0 +1,98 @@
+package au.org.ala.pipelines.beam;
+
+import au.org.ala.pipelines.options.DistributionOutlierPipelineOptions;
+import au.org.ala.pipelines.transforms.DistributionOutlierTransform;
+import au.org.ala.pipelines.util.VersionInfo;
+import au.org.ala.utils.ALAFsUtils;
+import au.org.ala.utils.CombinedYamlConfiguration;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.io.AvroIO;
+import org.apache.beam.sdk.transforms.*;
+import org.apache.beam.sdk.values.PCollection;
+import org.gbif.pipelines.common.beam.options.PipelinesOptionsFactory;
+import org.gbif.pipelines.io.avro.*;
+import org.slf4j.MDC;
+
+/**
+ * A pipeline that calculate distance to the expert distribution layers (EDL)
+ *
+ * <p>distanceOutOfELD: 0 -> inside of EDL, -1: -> No EDLs. >0 -> out of EDL
+ *
+ * <p>Example: java au.org.ala.pipelines.beam.DistributionOutlierPipeline
+ * --config=/data/la-pipelines/config/la-pipelines.yaml --fsPath=/data
+ *
+ * <p>Running with Jar java
+ * -Dlog4j.configuration=file://../pipelines/src/main/resources/log4j-colorized.properties
+ * -Dlog4j.configurationFile=file://../pipelines/src/main/resources/log4j-colorized.properties -cp
+ * ../pipelines/target/pipelines-2.10.0-SNAPSHOT-shaded.jar
+ * au.org.ala.pipelines.beam.DistributionOutlierPipeline
+ * --config=/data/la-pipelines/config/la-pipelines.yaml,la-pipelines-local.yaml --fsPath=/data
+ */
+@Slf4j
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class DistributionOutlierPipeline {
+
+  public static void main(String[] args) throws Exception {
+    VersionInfo.print();
+    CombinedYamlConfiguration conf = new CombinedYamlConfiguration(args);
+    String[] combinedArgs = conf.toArgs("general", "outlier");
+
+    DistributionOutlierPipelineOptions options =
+        PipelinesOptionsFactory.create(DistributionOutlierPipelineOptions.class, combinedArgs);
+    MDC.put("datasetId", options.getDatasetId());
+    MDC.put("attempt", options.getAttempt().toString());
+    MDC.put("step", "OUTLIER");
+    PipelinesOptionsFactory.registerHdfs(options);
+    run(options);
+    System.exit(0);
+  }
+
+  public static void run(DistributionOutlierPipelineOptions options) throws Exception {
+
+    // Create output path
+    // default: {fsPath}/pipelines-outlier
+    // or {fsPath}/pipelines-outlier/{datasetId}
+    String outputPath = ALAFsUtils.buildPathOutlierUsingTargetPath(options);
+
+    log.info("Adding step 1: Collecting index records");
+    Pipeline p = Pipeline.create(options);
+
+    PCollection<IndexRecord> indexRecords = ALAFsUtils.loadIndexRecords(options, p);
+
+    DistributionOutlierTransform distributionTransform =
+        new DistributionOutlierTransform(options.getBaseUrl());
+
+    log.info("Adding step 2: calculating outliers index");
+    PCollection<DistributionOutlierRecord> kvRecords =
+        indexRecords
+            .apply("Key by species", distributionTransform.toKv())
+            .apply("Grouping by species", GroupByKey.create())
+            .apply(
+                "Calculating outliers based on the species",
+                distributionTransform.calculateOutlier())
+            .apply("Flatten records", Flatten.iterables());
+
+    log.info("Adding step 3: writing to outliers");
+
+    kvRecords.apply(
+        "Write to file",
+        AvroIO.write(DistributionOutlierRecord.class)
+            .to(outputPath + "/outlier")
+            .withoutSharding()
+            .withSuffix(".avro"));
+
+    //    kvRecords
+    //        .apply("to String", distributionTransform.flatToString())
+    //        .apply(
+    //            "Write to text",
+    //            TextIO.write().to(outputPath+"/outlier").withoutSharding().withSuffix(".txt"));
+
+    log.info("Running the pipeline");
+    PipelineResult result = p.run();
+    result.waitUntilFinish();
+  }
+}