Wayang AgoraEO

pom.xml

@@ -483,9 +483,9 @@
                         <groupId>net.alchim31.maven</groupId>
                         <artifactId>scala-maven-plugin</artifactId>
                     </plugin>
-                    <!--                    <plugin>-->
-                    <!--                        <artifactId>maven-jar-plugin</artifactId>-->
-                    <!--                    </plugin>-->
+<!--                                        <plugin>-->
+<!--                                            <artifactId>maven-jar-plugin</artifactId>-->
+<!--                                        </plugin>-->
                 </plugins>
             </build>
         </profile>

wayang-commons/wayang-basic/pom.xml

@@ -61,6 +61,7 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
         </dependency>
+
     </dependencies>
 
 </project>

wayang-platforms/wayang-spark/code/main/resources/wayang-spark-defaults.properties

@@ -15,7 +15,7 @@
 # limitations under the License.
 #
 
-spark.master = local[1]
+spark.master = local[*]
 spark.app.name = Wayang App
 spark.ui.showConsoleProgress = false
 spark.driver.allowMultipleContexts=true

wayang-plugins/pom.xml

@@ -40,6 +40,7 @@
 
     <modules>
         <module>wayang-iejoin</module>
+        <module>wayang-agoraeo</module>
     </modules>
 
 

wayang-plugins/wayang-agoraeo/pom.xml

@@ -0,0 +1,80 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>wayang-plugins</artifactId>
+        <groupId>org.apache.wayang</groupId>
+        <version>0.6.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>wayang-agoraeo</artifactId>
+    <version>0.6.1-SNAPSHOT</version>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.gdal</groupId>
+                <artifactId>gdal</artifactId>
+                <version>3.6.0</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-core</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-basic</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-java</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.wayang</groupId>
+            <artifactId>wayang-spark_2.12</artifactId>
+            <version>0.6.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>3.2.4</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client</artifactId>
+            <version>3.1.2</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_2.13</artifactId>
+            <version>3.3.0</version>
+            <scope>provided</scope>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/org.gdal/gdal -->
+        <dependency>
+            <groupId>org.gdal</groupId>
+            <artifactId>gdal</artifactId>
+            <version>3.7.0</version>
+        </dependency>
+
+    </dependencies>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+</project>

wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java

@@ -0,0 +1,103 @@
+/*
+ * 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;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.wayang.agoraeo.operators.basic.Sen2CorWrapper;
+import org.apache.wayang.agoraeo.operators.basic.SentinelSource;
+import org.apache.wayang.agoraeo.sentinel.Mirror;
+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.plan.wayangplan.WayangPlan;
+import org.apache.wayang.core.util.ReflectionUtils;
+import org.apache.wayang.java.Java;
+import org.apache.wayang.java.platform.JavaPlatform;
+import org.apache.wayang.spark.Spark;
+import org.apache.wayang.spark.platform.SparkPlatform;
+
+public class Main {
+    public static void main(String[] args) {
+
+        WayangContext wayangContext = new WayangContext();
+        wayangContext.register(Java.basicPlugin());
+        wayangContext.register(Spark.basicPlugin());
+        wayangContext.register(WayangAgoraEO.plugin());
+
+        Configuration config = wayangContext.getConfiguration();
+        config.load(ReflectionUtils.loadResource(WayangAgoraEO.DEFAULT_CONFIG_FILE));
+
+        String sen2cor = config.getStringProperty("org.apache.wayang.agoraeo.sen2cor.location");
+        String l2a_images_folder = config.getStringProperty("org.apache.wayang.agoraeo.images.l2a");
+
+        System.out.println("Running AgoraEO!");
+
+        // TODO: Read all from config file
+        Mirror m1 = new Mirror("https://scihub.copernicus.eu/dhus", "rpardomeza", "12c124ccb2");
+        Mirror m2 = new Mirror("https://colhub.met.no", "rpardomeza", "12c124ccb2");
+
+        List<Mirror> mirrors = Arrays.asList(m1,m2);
+
+        DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyyMMdd-HH:mm:ss");
+        LocalDateTime now = LocalDateTime.now();
+        String sen2corlog = now.format(dtf);
+
+        WayangPlan w = alternative2WayangPlan(
+                mirrors,
+                sen2cor,
+                l2a_images_folder,
+                "file:///Users/rodrigopardomeza/tu-berlin/agoraeo/agoraeo/sen2cor_logs/"+sen2corlog
+        );
+
+        wayangContext.execute(w, ReflectionUtils.getDeclaringJar(Main.class), ReflectionUtils.getDeclaringJar(JavaPlatform.class), ReflectionUtils.getDeclaringJar(SparkPlatform.class));
+
+    }
+
+    public static WayangPlan alternative2WayangPlan(
+            List<Mirror> mirrors,
+            String sen2cor,
+            String l2a_location,
+            String outputFileUrl
+    ) {
+
+        SentinelSource source = new SentinelSource()
+            .setFrom("NOW-30DAY")
+            .setTo("NOW")
+            .setOrder(Arrays.asList("33UWP", "32VNM"))
+            .setMirrors(mirrors)
+        ;
+
+        Sen2CorWrapper toL2A = new Sen2CorWrapper(sen2cor, l2a_location);
+
+        /* TODO: BigEarthNet Pipeline */
+
+        TextFileSink<String> sink = new TextFileSink<>(outputFileUrl, String.class);
+
+        source.connectTo(0,toL2A,0);
+        toL2A.connectTo(0,sink,0);
+
+
+        return new WayangPlan(sink);
+    }
+
+}