You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@wayang.apache.org by rp...@apache.org on 2022/12/12 19:54:48 UTC
[incubator-wayang] branch WAYANG-agoraeo updated (ecd18a30 -> 37351e3b)
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a change to branch WAYANG-agoraeo
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
from ecd18a30 Config fix
new b11fc8a5 Sen2Cor wrapper class
new 37351e3b New SentinelSource Definition plus Simplified query
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/java/org/apache/wayang/agoraeo/Main.java | 320 ++++++---------------
.../apache/wayang/agoraeo/mappings/Mappings.java | 4 +-
...urceMapping.java => Sen2CorWrapperMapping.java} | 15 +-
.../mappings/java/SentinelSourceMapping.java | 6 +-
.../agoraeo/operators/basic/Sen2CorWrapper.java | 28 ++
.../agoraeo/operators/basic/SentinelSource.java | 75 ++++-
.../agoraeo/operators/java/JavaSen2CorWrapper.java | 118 ++++++++
.../agoraeo/operators/java/JavaSentinelSource.java | 10 +-
8 files changed, 313 insertions(+), 263 deletions(-)
copy wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/{SentinelSourceMapping.java => Sen2CorWrapperMapping.java} (71%)
create mode 100644 wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/Sen2CorWrapper.java
create mode 100644 wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSen2CorWrapper.java
[incubator-wayang] 01/02: Sen2Cor wrapper class
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch WAYANG-agoraeo
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit b11fc8a5cb55ec812ad2501cd8ba36aa68d0ef4c
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Mon Dec 12 19:40:53 2022 +0100
Sen2Cor wrapper class
---
.../main/java/org/apache/wayang/agoraeo/Main.java | 169 ++-------------------
.../apache/wayang/agoraeo/mappings/Mappings.java | 4 +-
.../mappings/java/Sen2CorWrapperMapping.java | 43 ++++++
.../agoraeo/operators/basic/Sen2CorWrapper.java | 28 ++++
.../agoraeo/operators/java/JavaSen2CorWrapper.java | 118 ++++++++++++++
5 files changed, 208 insertions(+), 154 deletions(-)
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
index 2dd5472f..3ca32ddd 100644
--- a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
@@ -8,6 +8,7 @@ import java.util.stream.Stream;
import org.apache.wayang.agoraeo.iterators.FileIteratorSentinelDownload;
import org.apache.wayang.agoraeo.iterators.IteratorSentinelDownload;
import org.apache.wayang.agoraeo.iterators.StringIteratorSentinelDownload;
+import org.apache.wayang.agoraeo.operators.basic.Sen2CorWrapper;
import org.apache.wayang.agoraeo.operators.basic.SentinelSource;
import org.apache.wayang.basic.data.Tuple2;
import org.apache.wayang.basic.operators.*;
@@ -32,115 +33,29 @@ public class Main {
wayangContext.register(Java.basicPlugin());
wayangContext.register(WayangAgoraEO.javaPlugin());
- Configuration config = wayangContext.getConfiguration();//.fork(this.name);
-
+ Configuration config = wayangContext.getConfiguration();
config.load(ReflectionUtils.loadResource(WayangAgoraEO.DEFAULT_CONFIG_FILE));
- String path_python = "";
- if (config.getOptionalStringProperty("org.apache.wayang.agoraeo.python.location").isPresent()) {
- path_python = config.getOptionalStringProperty("org.apache.wayang.agoraeo.python.location").get();
- } else {
- throw new WayangException("No Python Interpreter to run Minimal Download");
- }
-
- String downloader_location = "";
- if (config.getOptionalStringProperty("org.apache.wayang.agoraeo.minimaldownload.location").isPresent()) {
- downloader_location = config.getOptionalStringProperty("org.apache.wayang.agoraeo.minimaldownload.location").get();
- } else {
- throw new WayangException("No defined Minimal Download path");
- }
-
- String sen2cor = "";
- if (config.getOptionalStringProperty("org.apache.wayang.agoraeo.sen2cor.location").isPresent()) {
- sen2cor = config.getOptionalStringProperty("org.apache.wayang.agoraeo.sen2cor.location").get();
- } else {
- throw new WayangException("No defined Minimal Download path");
- }
-
- String l2a_images_folder = "";
- if (config.getOptionalStringProperty("org.apache.wayang.agoraeo.images.l2a").isPresent()) {
- l2a_images_folder = config.getOptionalStringProperty("org.apache.wayang.agoraeo.images.l2a").get();
- } else {
- throw new WayangException("No Python Interpreter to run Minimal Download");
- }
+ String path_python = config.getStringProperty("org.apache.wayang.agoraeo.python.location");
+ String downloader_location = config.getStringProperty("org.apache.wayang.agoraeo.minimaldownload.location");
+ String sen2cor = config.getStringProperty("org.apache.wayang.agoraeo.sen2cor.location");
+ String l2a_images_folder = config.getStringProperty("org.apache.wayang.agoraeo.images.l2a");
+ String user = config.getStringProperty("org.apache.wayang.agoraeo.user");
+ String pass = config.getStringProperty("org.apache.wayang.agoraeo.pass");
+ String mirror = config.getStringProperty("org.apache.wayang.agoraeo.mirror");
- String user = "";
- if (config.getOptionalStringProperty("org.apache.wayang.agoraeo.user").isPresent()) {
- user = config.getOptionalStringProperty("org.apache.wayang.agoraeo.user").get();
- } else {
- throw new WayangException("No Python Interpreter to run Minimal Download");
- }
+ System.out.println("Running AgoraEO!");
- String pass = "";
- if (config.getOptionalStringProperty("org.apache.wayang.agoraeo.pass").isPresent()) {
- pass = config.getOptionalStringProperty("org.apache.wayang.agoraeo.pass").get();
- } else {
- throw new WayangException("No Python Interpreter to run Minimal Download");
- }
-
- System.out.println("Hello AgoraEO!");
-
- String order = " --url https://scihub.copernicus.eu/dhus --from NOW-30DAY --to NOW --order 32VNM";
+ String order = "--from NOW-30DAY --to NOW --order 34SGH";
String cmd =
- path_python + " " + downloader_location + " --user " + user + " --password " + pass + order;
-
+ path_python + " " + downloader_location + " --url " + mirror + " --user " + user + " --password " + pass + " " + order;
- System.out.println(path_python);
- System.out.println(downloader_location);
- System.out.println(l2a_images_folder);
- System.out.println(cmd);
- WayangPlan w = alternative2WayangPlan(cmd, sen2cor, l2a_images_folder, "");
+ WayangPlan w = alternative2WayangPlan(cmd, sen2cor, l2a_images_folder, "file:///Users/rodrigopardomeza/files/sen2cor-output-agoraeo.txt");
wayangContext.execute(w, ReflectionUtils.getDeclaringJar(Main.class), ReflectionUtils.getDeclaringJar(JavaPlatform.class));
}
- public static WayangPlan alternativeWayangPlan(
- String cmd,
- String outputFileUrl
- ) {
- IteratorSentinelDownload<File> iter = new FileIteratorSentinelDownload("Sentinel 2 - API", cmd);
-
- SentinelSource<File> source = new SentinelSource<>(iter, File.class);
-
- /*TODO Replace String to call for Sen2Cor and retrieve output of the command*/
- MapOperator<File, String> files_names = new MapOperator<>(
- t -> t.getAbsolutePath(),
- File.class,
- String.class
- );
-
-// FlatMapOperator<File, String> files_names = new FlatMapOperator<>(
-// t -> {
-// try {
-// String command = "~/Downloads/Sen2Cor-02.10.01-Darwin64/bin/L2A_Process\n" +
-// "~/tu-berlin/agoraeo/images/33UVT/S2B_MSIL1C_20221119T101229_N0400_R022_T33UVT_20221119T104925.SAFE\n" +
-// "--output_dir ~/tu-berlin/agoraeo/images/33UVT/L2A/";
-// Process process = Runtime.getRuntime().exec(command);
-// Iterator<String> input = new BufferedReader(
-// new InputStreamReader(
-// process.getInputStream()
-// )
-// ).lines().iterator();
-// return () -> input;
-// } catch (IOException e) {
-// throw new RuntimeException(e);
-// }
-// },
-// File.class,
-// String.class
-// );
-
-
- TextFileSink<String> sink = new TextFileSink<>(outputFileUrl, String.class);
-
- source.connectTo(0,files_names,0);
- files_names.connectTo(0,sink,0);
-
-
- return new WayangPlan(sink);
- }
-
public static WayangPlan alternative2WayangPlan(
String cmd,
String sen2cor,
@@ -151,65 +66,13 @@ public class Main {
SentinelSource<String> source = new SentinelSource<>(iter, String.class);
- FlatMapOperator<String, String> files_names = new FlatMapOperator<>(
- t -> {
- try {
- String command = sen2cor + " " +
- t + " " +
- " --output_dir " + l2a_location;
- Process process = Runtime.getRuntime().exec(command);
- Iterator<String> input = new BufferedReader(
- new InputStreamReader(
- process.getInputStream()
- )
- ).lines().iterator();
- return () -> input;
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- },
- String.class,
- String.class
- );
-
+ Sen2CorWrapper toL2A = new Sen2CorWrapper(sen2cor, l2a_location);
TextFileSink<String> sink = new TextFileSink<>(outputFileUrl, String.class);
- source.connectTo(0,files_names,0);
- files_names.connectTo(0,sink,0);
-
-
- return new WayangPlan(sink);
- }
-
- public static WayangPlan basicWayangPlan(
- String inputFileUrl,
- String outputFileUrl
-
- ) {
-
- TextFileSource textFileSource = new TextFileSource(inputFileUrl);
-
- FlatMapOperator<String, String> flatMapOperator = new FlatMapOperator<>(
- new FlatMapDescriptor<>(line -> Arrays.asList(line.split("\\W+")),
- String.class,
- String.class,
- new ProbabilisticDoubleInterval(100, 10000, 0.8)
- )
- );
-
- flatMapOperator.setName("Split words");
-
- FilterOperator<String> filterOperator = new FilterOperator<>(str -> !str.isEmpty(), String.class);
- filterOperator.setName("Filter empty words");
-
- // write results to a sink
- TextFileSink<String> sink = new TextFileSink<String>(outputFileUrl, String.class);
- sink.setName("Write result");
+ source.connectTo(0,toL2A,0);
+ toL2A.connectTo(0,sink,0);
- textFileSource.connectTo(0, flatMapOperator, 0);
- flatMapOperator.connectTo(0, filterOperator, 0);
- filterOperator.connectTo(0, sink, 0);
return new WayangPlan(sink);
}
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/Mappings.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/Mappings.java
index 60c86db1..9c50e1c7 100644
--- a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/Mappings.java
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/Mappings.java
@@ -1,5 +1,6 @@
package org.apache.wayang.agoraeo.mappings;
+import org.apache.wayang.agoraeo.mappings.java.Sen2CorWrapperMapping;
import org.apache.wayang.agoraeo.mappings.java.SentinelSourceMapping;
import org.apache.wayang.core.mapping.Mapping;
@@ -9,7 +10,8 @@ import java.util.Collection;
public class Mappings {
public static Collection<Mapping> BASIC_MAPPINGS = Arrays.asList(
- new SentinelSourceMapping()
+ new SentinelSourceMapping(),
+ new Sen2CorWrapperMapping()
);
}
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/Sen2CorWrapperMapping.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/Sen2CorWrapperMapping.java
new file mode 100644
index 00000000..01965fdd
--- /dev/null
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/Sen2CorWrapperMapping.java
@@ -0,0 +1,43 @@
+package org.apache.wayang.agoraeo.mappings.java;
+
+import org.apache.wayang.agoraeo.operators.basic.Sen2CorWrapper;
+import org.apache.wayang.agoraeo.operators.basic.SentinelSource;
+import org.apache.wayang.agoraeo.operators.java.JavaSen2CorWrapper;
+import org.apache.wayang.agoraeo.operators.java.JavaSentinelSource;
+import org.apache.wayang.basic.operators.FlatMapOperator;
+import org.apache.wayang.core.mapping.*;
+import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.java.platform.JavaPlatform;
+
+import java.util.Collection;
+import java.util.Collections;
+
+public class Sen2CorWrapperMapping implements Mapping {
+
+ @Override
+ public Collection<PlanTransformation> getTransformations() {
+ return Collections.singleton(new PlanTransformation(
+ this.createSubplanPattern(),
+ this.createReplacementSubplanFactory(),
+ JavaPlatform.getInstance()
+ ));
+ }
+
+ private SubplanPattern createSubplanPattern() {
+ final OperatorPattern operatorPattern = new OperatorPattern(
+ "sen2cor",
+ new Sen2CorWrapper(
+ null,
+ null
+ ),
+ false
+ );
+ return SubplanPattern.createSingleton(operatorPattern);
+ }
+
+ private ReplacementSubplanFactory createReplacementSubplanFactory() {
+ return new ReplacementSubplanFactory.OfSingleOperators<Sen2CorWrapper>(
+ (matchedOperator, epoch) -> new JavaSen2CorWrapper(matchedOperator).at(epoch)
+ );
+ }
+}
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/Sen2CorWrapper.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/Sen2CorWrapper.java
new file mode 100644
index 00000000..b5d425c3
--- /dev/null
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/Sen2CorWrapper.java
@@ -0,0 +1,28 @@
+package org.apache.wayang.agoraeo.operators.basic;
+
+import org.apache.wayang.core.plan.wayangplan.UnaryToUnaryOperator;
+import org.apache.wayang.core.types.BasicDataUnitType;
+import org.apache.wayang.core.types.DataSetType;
+
+public class Sen2CorWrapper
+ extends UnaryToUnaryOperator<String, String>
+{
+ public String sen2cor;
+ public String l2a_location;
+
+ public Sen2CorWrapper(String sen2cor, String l2a_location){
+ super(
+ DataSetType.createDefault(BasicDataUnitType.createBasic(String.class)),
+ DataSetType.createDefault(BasicDataUnitType.createBasic(String.class)),
+ false
+ );
+ this.sen2cor = sen2cor;
+ this.l2a_location = l2a_location;
+ }
+
+ public Sen2CorWrapper(UnaryToUnaryOperator<String, String> that) {
+ super(that);
+ }
+
+
+}
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSen2CorWrapper.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSen2CorWrapper.java
new file mode 100644
index 00000000..7e2fa90c
--- /dev/null
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSen2CorWrapper.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.agoraeo.operators.java;
+
+import org.apache.wayang.agoraeo.operators.basic.Sen2CorWrapper;
+import org.apache.wayang.core.optimizer.OptimizationContext.OperatorContext;
+import org.apache.wayang.core.plan.wayangplan.ExecutionOperator;
+import org.apache.wayang.core.platform.ChannelDescriptor;
+import org.apache.wayang.core.platform.ChannelInstance;
+import org.apache.wayang.core.platform.lineage.ExecutionLineageNode;
+import org.apache.wayang.core.util.Tuple;
+import org.apache.wayang.java.channels.CollectionChannel;
+import org.apache.wayang.java.channels.JavaChannelInstance;
+import org.apache.wayang.java.channels.StreamChannel;
+import org.apache.wayang.java.execution.JavaExecutor;
+import org.apache.wayang.java.operators.JavaExecutionOperator;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.StreamSupport;
+
+//TODO add the documentation and add the Profile Estimator
+public class JavaSen2CorWrapper
+ extends Sen2CorWrapper
+ implements JavaExecutionOperator {
+
+
+ public JavaSen2CorWrapper(String sen2cor, String l2a_location) {
+ super(sen2cor, l2a_location);
+ }
+
+ @Override
+ public List<ChannelDescriptor> getSupportedInputChannels(int index) {
+ assert index <= this.getNumInputs() || (index == 0 && this.getNumInputs() == 0);
+ if (this.getInput(index).isBroadcast()) return Collections.singletonList(CollectionChannel.DESCRIPTOR);
+ return Arrays.asList(CollectionChannel.DESCRIPTOR, StreamChannel.DESCRIPTOR);
+ }
+
+ @Override
+ public List<ChannelDescriptor> getSupportedOutputChannels(int index) {
+ assert index <= this.getNumOutputs() || (index == 0 && this.getNumOutputs() == 0);
+ return Collections.singletonList(StreamChannel.DESCRIPTOR);
+ }
+
+ @Override
+ public Tuple<Collection<ExecutionLineageNode>, Collection<ChannelInstance>> evaluate(ChannelInstance[] inputs, ChannelInstance[] outputs, JavaExecutor javaExecutor, OperatorContext operatorContext) {
+
+ assert inputs.length == this.getNumInputs();
+ assert outputs.length == this.getNumOutputs();
+
+ final Function<String, Iterable<String>> flatmapFunction =
+ t -> {
+ System.out.println("sen2cor");
+ System.out.println(sen2cor);
+ System.out.println("l2a_location");
+ System.out.println(l2a_location);
+ try {
+ String command = sen2cor + " " +
+ t + " " +
+ " --output_dir " + l2a_location;
+ Process process = Runtime.getRuntime().exec(command);
+ Iterator<String> input = new BufferedReader(
+ new InputStreamReader(
+ process.getInputStream()
+ )
+ ).lines().iterator();
+ return () -> input;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ };
+// javaExecutor.getCompiler().compile(this.functionDescriptor);
+
+ JavaExecutor.openFunction(this, flatmapFunction, inputs, operatorContext);
+
+ ((StreamChannel.Instance) outputs[0]).accept(
+ ((JavaChannelInstance) inputs[0]).<String>provideStream().flatMap(dataQuantum ->
+ StreamSupport.stream(
+ Spliterators.spliteratorUnknownSize(
+ flatmapFunction.apply(dataQuantum).iterator(),
+ Spliterator.ORDERED),
+ false
+ )
+ )
+ );
+
+ return ExecutionOperator.modelLazyExecution(inputs, outputs, operatorContext);
+ }
+
+ public JavaSen2CorWrapper(Sen2CorWrapper that) {
+ super(that);
+ this.sen2cor = that.sen2cor;
+ this.l2a_location = that.l2a_location;
+ }
+
+ protected ExecutionOperator createCopy() {
+ return new JavaSen2CorWrapper(this.sen2cor, this.l2a_location);
+ }
+}
[incubator-wayang] 02/02: New SentinelSource Definition plus Simplified query
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch WAYANG-agoraeo
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 37351e3bb73f15f7f9bd05cdd4d2759d116ab712
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Mon Dec 12 20:54:34 2022 +0100
New SentinelSource Definition plus Simplified query
---
.../main/java/org/apache/wayang/agoraeo/Main.java | 171 +++++++++------------
.../mappings/java/SentinelSourceMapping.java | 6 +-
.../agoraeo/operators/basic/SentinelSource.java | 75 +++++++--
.../agoraeo/operators/java/JavaSentinelSource.java | 10 +-
4 files changed, 149 insertions(+), 113 deletions(-)
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
index 3ca32ddd..f9004a17 100644
--- a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
@@ -1,27 +1,11 @@
package org.apache.wayang.agoraeo;
-import java.io.*;
-import java.nio.file.Files;
-import java.util.*;
-import java.util.stream.Stream;
-
-import org.apache.wayang.agoraeo.iterators.FileIteratorSentinelDownload;
-import org.apache.wayang.agoraeo.iterators.IteratorSentinelDownload;
-import org.apache.wayang.agoraeo.iterators.StringIteratorSentinelDownload;
import org.apache.wayang.agoraeo.operators.basic.Sen2CorWrapper;
import org.apache.wayang.agoraeo.operators.basic.SentinelSource;
-import org.apache.wayang.basic.data.Tuple2;
import org.apache.wayang.basic.operators.*;
import org.apache.wayang.core.api.Configuration;
import org.apache.wayang.core.api.WayangContext;
-import org.apache.wayang.core.api.exception.WayangException;
-import org.apache.wayang.core.function.FlatMapDescriptor;
-import org.apache.wayang.core.function.ReduceDescriptor;
-import org.apache.wayang.core.function.TransformationDescriptor;
-import org.apache.wayang.core.optimizer.ProbabilisticDoubleInterval;
import org.apache.wayang.core.plan.wayangplan.WayangPlan;
-import org.apache.wayang.core.types.DataSetType;
-import org.apache.wayang.core.types.DataUnitType;
import org.apache.wayang.core.util.ReflectionUtils;
import org.apache.wayang.java.Java;
import org.apache.wayang.java.platform.JavaPlatform;
@@ -36,38 +20,32 @@ public class Main {
Configuration config = wayangContext.getConfiguration();
config.load(ReflectionUtils.loadResource(WayangAgoraEO.DEFAULT_CONFIG_FILE));
- String path_python = config.getStringProperty("org.apache.wayang.agoraeo.python.location");
- String downloader_location = config.getStringProperty("org.apache.wayang.agoraeo.minimaldownload.location");
String sen2cor = config.getStringProperty("org.apache.wayang.agoraeo.sen2cor.location");
String l2a_images_folder = config.getStringProperty("org.apache.wayang.agoraeo.images.l2a");
- String user = config.getStringProperty("org.apache.wayang.agoraeo.user");
- String pass = config.getStringProperty("org.apache.wayang.agoraeo.pass");
- String mirror = config.getStringProperty("org.apache.wayang.agoraeo.mirror");
System.out.println("Running AgoraEO!");
- String order = "--from NOW-30DAY --to NOW --order 34SGH";
- String cmd =
- path_python + " " + downloader_location + " --url " + mirror + " --user " + user + " --password " + pass + " " + order;
+ String order = "--from NOW-30DAY --to NOW --order 33UUU";
- WayangPlan w = alternative2WayangPlan(cmd, sen2cor, l2a_images_folder, "file:///Users/rodrigopardomeza/files/sen2cor-output-agoraeo.txt");
+ WayangPlan w = alternative2WayangPlan(order, sen2cor, l2a_images_folder, "file:///Users/rodrigopardomeza/files/sen2cor-output-agoraeo.txt");
wayangContext.execute(w, ReflectionUtils.getDeclaringJar(Main.class), ReflectionUtils.getDeclaringJar(JavaPlatform.class));
}
public static WayangPlan alternative2WayangPlan(
- String cmd,
+ String order,
String sen2cor,
String l2a_location,
String outputFileUrl
) {
- IteratorSentinelDownload<String> iter = new StringIteratorSentinelDownload("Sentinel 2 - API", cmd);
- SentinelSource<String> source = new SentinelSource<>(iter, String.class);
+ SentinelSource source = new SentinelSource(order);
Sen2CorWrapper toL2A = new Sen2CorWrapper(sen2cor, l2a_location);
+ /* TODO: BigEarthNet Pipeline */
+
TextFileSink<String> sink = new TextFileSink<>(outputFileUrl, String.class);
source.connectTo(0,toL2A,0);
@@ -77,73 +55,76 @@ public class Main {
return new WayangPlan(sink);
}
- public static WayangPlan createWayangPlan(
- String cmd,
- String outputFileUrl
- ) {
- IteratorSentinelDownload<File> iter = new FileIteratorSentinelDownload("Sentinel 2 - API", cmd);
-
- /* Might replay the name of the downloaded file */
- SentinelSource<File> source = new SentinelSource<>(iter, File.class);
-
- FlatMapOperator<File, String> files_lines = new FlatMapOperator<>(
- t -> {
- try {
- final InputStream inputStream = Files.newInputStream(t.toPath());
- Stream<String> rough_lines = new BufferedReader(new InputStreamReader(inputStream)).lines();
- return rough_lines::iterator;
- } catch (IOException e) {
- throw new WayangException(String.format("Reading %s failed.", t), e);
- }
- },
- File.class,
- String.class
- );
- files_lines.setName("files giving lines");
-
- FlatMapOperator<String, String> words = new FlatMapOperator<>(
- line -> Arrays.asList(line.split("\\W+")),
- String.class,
- String.class
- );
- words.setName("words");
-
- // for each word transform it to lowercase and output a key-value pair (word, 1)
- MapOperator<String, Tuple2<String, Integer>> mapOperator = new MapOperator<>(
- new TransformationDescriptor<>(word -> new Tuple2<>(word.toLowerCase(), 1),
- DataUnitType.createBasic(String.class),
- DataUnitType.createBasicUnchecked(Tuple2.class)
- ), DataSetType.createDefault(String.class),
- DataSetType.createDefaultUnchecked(Tuple2.class)
- );
- mapOperator.setName("To lower case, add counter");
-
-
- // groupby the key (word) and add up the values (frequency)
- ReduceByOperator<Tuple2<String, Integer>, String> reduceByOperator = new ReduceByOperator<>(
- new TransformationDescriptor<>(pair -> pair.field0,
- DataUnitType.createBasicUnchecked(Tuple2.class),
- DataUnitType.createBasic(String.class)), new ReduceDescriptor<>(
- ((a, b) -> {
- a.field1 += b.field1;
- return a;
- }), DataUnitType.createGroupedUnchecked(Tuple2.class),
- DataUnitType.createBasicUnchecked(Tuple2.class)
- ), DataSetType.createDefaultUnchecked(Tuple2.class)
- );
- reduceByOperator.setName("Add counters");
-
- TextFileSink<Tuple2> sink = new TextFileSink<>(outputFileUrl, Tuple2.class);
- sink.setName("Collect result");
-
-
- source.connectTo(0, files_lines, 0);
- files_lines.connectTo(0, words,0);
- words.connectTo(0, mapOperator, 0);
- mapOperator.connectTo(0, reduceByOperator, 0);
- reduceByOperator.connectTo(0, sink, 0);
- return new WayangPlan(sink);
- }
+
+// public static WayangPlan createWayangPlan(
+// String cmd,
+// String outputFileUrl
+// ) {
+//
+// IteratorSentinelDownload<File> iter = new FileIteratorSentinelDownload("Sentinel 2 - API", cmd);
+//
+// /* Might replay the name of the downloaded file */
+// SentinelSource<File> source = new SentinelSource<>(iter, File.class);
+//
+// FlatMapOperator<File, String> files_lines = new FlatMapOperator<>(
+// t -> {
+// try {
+// final InputStream inputStream = Files.newInputStream(t.toPath());
+// Stream<String> rough_lines = new BufferedReader(new InputStreamReader(inputStream)).lines();
+// return rough_lines::iterator;
+// } catch (IOException e) {
+// throw new WayangException(String.format("Reading %s failed.", t), e);
+// }
+// },
+// File.class,
+// String.class
+// );
+// files_lines.setName("files giving lines");
+//
+// FlatMapOperator<String, String> words = new FlatMapOperator<>(
+// line -> Arrays.asList(line.split("\\W+")),
+// String.class,
+// String.class
+// );
+// words.setName("words");
+//
+// // for each word transform it to lowercase and output a key-value pair (word, 1)
+// MapOperator<String, Tuple2<String, Integer>> mapOperator = new MapOperator<>(
+// new TransformationDescriptor<>(word -> new Tuple2<>(word.toLowerCase(), 1),
+// DataUnitType.createBasic(String.class),
+// DataUnitType.createBasicUnchecked(Tuple2.class)
+// ), DataSetType.createDefault(String.class),
+// DataSetType.createDefaultUnchecked(Tuple2.class)
+// );
+// mapOperator.setName("To lower case, add counter");
+//
+//
+// // groupby the key (word) and add up the values (frequency)
+// ReduceByOperator<Tuple2<String, Integer>, String> reduceByOperator = new ReduceByOperator<>(
+// new TransformationDescriptor<>(pair -> pair.field0,
+// DataUnitType.createBasicUnchecked(Tuple2.class),
+// DataUnitType.createBasic(String.class)), new ReduceDescriptor<>(
+// ((a, b) -> {
+// a.field1 += b.field1;
+// return a;
+// }), DataUnitType.createGroupedUnchecked(Tuple2.class),
+// DataUnitType.createBasicUnchecked(Tuple2.class)
+// ), DataSetType.createDefaultUnchecked(Tuple2.class)
+// );
+// reduceByOperator.setName("Add counters");
+//
+// TextFileSink<Tuple2> sink = new TextFileSink<>(outputFileUrl, Tuple2.class);
+// sink.setName("Collect result");
+//
+//
+// source.connectTo(0, files_lines, 0);
+// files_lines.connectTo(0, words,0);
+// words.connectTo(0, mapOperator, 0);
+// mapOperator.connectTo(0, reduceByOperator, 0);
+// reduceByOperator.connectTo(0, sink, 0);
+//
+// return new WayangPlan(sink);
+// }
}
\ No newline at end of file
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/SentinelSourceMapping.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/SentinelSourceMapping.java
index 134b5677..90504838 100644
--- a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/SentinelSourceMapping.java
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/mappings/java/SentinelSourceMapping.java
@@ -3,7 +3,6 @@ package org.apache.wayang.agoraeo.mappings.java;
import org.apache.wayang.agoraeo.operators.basic.SentinelSource;
import org.apache.wayang.agoraeo.operators.java.JavaSentinelSource;
import org.apache.wayang.core.mapping.*;
-import org.apache.wayang.core.types.DataSetType;
import org.apache.wayang.java.platform.JavaPlatform;
import java.util.Collection;
@@ -23,9 +22,8 @@ public class SentinelSourceMapping implements Mapping {
private SubplanPattern createSubplanPattern() {
final OperatorPattern operatorPattern = new OperatorPattern(
"source",
- new SentinelSource<>(
- null,
- DataSetType.none().getDataUnitType().getTypeClass()
+ new SentinelSource(
+ ""
),
false
);
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/SentinelSource.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/SentinelSource.java
index 18987a8f..eea78064 100644
--- a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/SentinelSource.java
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/basic/SentinelSource.java
@@ -1,31 +1,88 @@
package org.apache.wayang.agoraeo.operators.basic;
+import org.apache.commons.io.IOUtils;
+import org.apache.wayang.agoraeo.WayangAgoraEO;
+import org.apache.wayang.agoraeo.iterators.StringIteratorSentinelDownload;
+import org.apache.wayang.core.api.exception.WayangException;
import org.apache.wayang.core.plan.wayangplan.UnarySource;
import org.apache.wayang.core.types.DataSetType;
+import org.apache.wayang.core.util.ReflectionUtils;
+import java.io.*;
import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
-public class SentinelSource<Type> extends UnarySource<Type> {
+public class SentinelSource extends UnarySource<String> {
- private Iterator<Type> iterator;
+ private Iterator<String> iterator;
/**
* Default construct
*
- * @param iterator {@link Iterator} Instance produced by this source
- * @param _class type of the data produced by the iterator
+ * @param order {@link String} what to do inside the command
*/
- public SentinelSource(Iterator<Type> iterator, Class<Type> _class){
- super(DataSetType.createDefault(_class));
- this.iterator = iterator;
+ public SentinelSource(String order){
+ super(DataSetType.createDefault(String.class));
+
+ InputStream str = ReflectionUtils.loadResource(WayangAgoraEO.DEFAULT_CONFIG_FILE);
+
+ String path_python = null;
+ String downloader_location = null;
+ String mirror = null;
+ String user = null;
+ String pass = null;
+
+ try {
+ final Properties properties = new Properties();
+ properties.load(str);
+ for (Map.Entry<Object, Object> propertyEntry : properties.entrySet()) {
+
+ final String key = propertyEntry.getKey().toString();
+ final String value = propertyEntry.getValue().toString();
+
+ if(Objects.equals(key, "org.apache.wayang.agoraeo.python.location")){
+ path_python = value;
+ }
+ if(Objects.equals(key, "org.apache.wayang.agoraeo.minimaldownload.location")){
+ downloader_location = value;
+ }
+ if(Objects.equals(key, "org.apache.wayang.agoraeo.mirror")){
+ mirror = value;
+ }
+ if(Objects.equals(key, "org.apache.wayang.agoraeo.user")){
+ user = value;
+ }
+ if(Objects.equals(key, "org.apache.wayang.agoraeo.pass")){
+ pass = value;
+ }
+
+ }
+ } catch (IOException e) {
+ throw new WayangException("Could not load configuration.", e);
+ } finally {
+ IOUtils.closeQuietly(str);
+ }
+
+ if(path_python == null || downloader_location == null || mirror == null || user == null || pass == null){
+ throw new WayangException("Missing inputs for Python Minimal Download");
+ }
+
+ String cmd = path_python + " " + downloader_location + " --url " + mirror + " --user " + user + " --password " + pass + " " + order;
+
+ this.iterator = new StringIteratorSentinelDownload("Sentinel 2 - API", cmd);
+
+// throw new WayangException("ABORT ON PURPOSE");
+
}
- public SentinelSource(SentinelSource<Type> that) {
+ public SentinelSource(SentinelSource that) {
super(that);
this.iterator = that.getIterator();
}
- public Iterator<Type> getIterator() {
+ public Iterator<String> getIterator() {
return this.iterator;
}
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSentinelSource.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSentinelSource.java
index 04d07ffc..0b7207be 100644
--- a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSentinelSource.java
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/operators/java/JavaSentinelSource.java
@@ -33,15 +33,15 @@ import java.util.*;
import java.util.stream.StreamSupport;
//TODO add the documentation and add the Profile Estimator
-public class JavaSentinelSource<Type>
- extends SentinelSource<Type>
+public class JavaSentinelSource
+ extends SentinelSource
implements JavaExecutionOperator {
- public JavaSentinelSource(Iterator<Type> iterator, Class<Type> _class) {
- super(iterator, _class);
+ public JavaSentinelSource(String iterator) {
+ super(iterator);
}
- public JavaSentinelSource(SentinelSource<Type> that) {
+ public JavaSentinelSource(SentinelSource that) {
super(that);
}