fetch updates

Merge branch 'master' into TB_Noaa_Buoy

Author: TomBlock

CHANGES.md

@@ -1,5 +1,8 @@
 ### Updates from version 1.5.7 to 1.5.8
 * added support for Windsat Coriolis data
+* added conda environment
+* extended workflow to run without scheduler on plain console
+* Added option to use different name for matchup dimension in post processing
 
 ### Updates from version 1.5.6 to 1.5.7
 * added support for SMOS L1C daily aggregated products

cems/src/main/python/monitor.py

@@ -1,7 +1,7 @@
 from pmonitor import PMonitor
 
 class Monitor:
-    def __init__(self, preconditions, usecase, hosts, types, log_dir, simulation):
+    def __init__(self, preconditions, usecase, hosts, types, log_dir, simulation, synchronous=False):
         """
 
         :type preconditions: list
@@ -10,8 +10,12 @@ def __init__(self, preconditions, usecase, hosts, types, log_dir, simulation):
         :type calls: list
         :type log_dir: str
         :type simulation: bool
+        :type synchronous: bool
         """
-        self.pm = PMonitor(preconditions, usecase, hosts, types, logdir=log_dir, simulation=simulation, polling="job_status_callback.sh")
+        if synchronous:
+            self.pm = PMonitor(preconditions, usecase, hosts, types, logdir=log_dir, simulation=simulation)
+        else:
+            self.pm = PMonitor(preconditions, usecase, hosts, types, logdir=log_dir, simulation=simulation, polling="job_status_callback.sh")
 
     def execute(self, job):
         """

cems/src/main/python/workflow.py

@@ -244,13 +244,14 @@ def _add_post_processing_preconditions(self, preconditions):
 
         return preconditions
 
-    def _get_monitor(self, hosts, calls, log_dir, simulation):
+    def _get_monitor(self, hosts, calls, log_dir, simulation, synchronous=False):
         """
 
         :type hosts: list
         :type calls: list
         :type log_dir: str
         :type simulation: bool
+        :type synchronous: bool
         :rtype : Monitor
         """
         preconditions = list()
@@ -262,7 +263,7 @@ def _get_monitor(self, hosts, calls, log_dir, simulation):
         # @todo 2 tb/tb do we need this 2016-03-29
         # self._add_obs_preconditions(preconditions)
         # self._add_smp_preconditions(preconditions)
-        return Monitor(preconditions, self.get_usecase(), hosts, calls, log_dir, simulation)
+        return Monitor(preconditions, self.get_usecase(), hosts, calls, log_dir, simulation, synchronous)
 
     def _next_year_start(self, date):
         """
@@ -366,16 +367,22 @@ def run_test_job(self, hosts, simulation=False, logdir='trace'):
 
         monitor.wait_for_completion()
 
-    def run_matchup(self, hosts, num_parallel_tasks, simulation=False, logdir='trace'):
+    def run_matchup(self, hosts, num_parallel_tasks, simulation=False, logdir='trace', synchronous=False):
         """
 
         :param hosts: list
         :param num_parallel_tasks: int
         :param simulation: bool
         :param logdir: str
+        :param synchronous: bool
         :return:
         """
-        monitor = self._get_monitor(hosts, [('matchup_start.sh', num_parallel_tasks)], logdir, simulation)
+        if synchronous:
+            runs_script = 'matchup_run.sh'
+        else:
+            runs_script = 'matchup_start.sh'
+
+        monitor = self._get_monitor(hosts, [(runs_script, num_parallel_tasks)], logdir, simulation, synchronous)
 
         sensors = self._get_sensor_pairs()
         for sensor_pair in sensors:
@@ -393,7 +400,7 @@ def run_matchup(self, hosts, num_parallel_tasks, simulation=False, logdir='trace
                 pre_condition = 'ingest-' + primary_name + '-' + start_string + '-' + end_string
                 post_condition = 'matchup-' + name + '-' + start_string + '-' + end_string + '-' + self.usecase_config
 
-                job = Job(job_name, 'matchup_start.sh', [pre_condition], [post_condition],
+                job = Job(job_name, runs_script, [pre_condition], [post_condition],
                           [start_string, end_string, self._get_config_dir(), self.usecase_config])
                 monitor.execute(job)
 

core/src/main/java/com/bc/fiduceo/geometry/s2/BcS2GeometryFactory.java

@@ -42,6 +42,7 @@ public BcS2GeometryFactory() {
     public Geometry parse(String wkt) {
         final Object geometry = s2WKTReader.read(wkt);
         if (geometry instanceof S2Polygon) {
+            //normalizePolygon((S2Polygon) geometry);
             return new BcS2Polygon(geometry);
         } else if (geometry instanceof S2Polyline) {
             return new BcS2LineString((S2Polyline) geometry);
@@ -53,6 +54,7 @@ public Geometry parse(String wkt) {
                 final ArrayList<Polygon> polygonList = new ArrayList<>();
                 final List<S2Polygon> googlePolygonList = (List<S2Polygon>) geometry;
                 for (S2Polygon googlePolygon : googlePolygonList) {
+                    //normalizePolygon(googlePolygon);
                     polygonList.add(new BcS2Polygon(googlePolygon));
                 }
                 return new BcS2MultiPolygon(polygonList);

core/src/main/java/com/bc/fiduceo/qc/FileMessages.java

@@ -0,0 +1,23 @@
+package com.bc.fiduceo.qc;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+class FileMessages {
+
+    private final HashMap<String, List<String>> messagesMap;
+
+    FileMessages() {
+        messagesMap = new HashMap<>();
+    }
+
+    HashMap<String, List<String>> getMessageMap() {
+        return messagesMap;
+    }
+
+    public void add(String fileName, String message) {
+        final List<String> messages = messagesMap.computeIfAbsent(fileName, k -> new ArrayList<>());
+        messages.add(message);
+    }
+}

core/src/main/java/com/bc/fiduceo/qc/MatchupAccumulator.java

@@ -0,0 +1,55 @@
+package com.bc.fiduceo.qc;
+
+import com.bc.fiduceo.util.TimeUtils;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.HashMap;
+
+class MatchupAccumulator {
+
+    private final Calendar utcCalendar;
+    private final HashMap<String, Integer> daysMap;
+    private int matchupSum;
+    private int fileCount;
+
+    public MatchupAccumulator() {
+        utcCalendar = TimeUtils.getUTCCalendar();
+        daysMap = new HashMap<>();
+        matchupSum = 0;
+        fileCount = 0;
+    }
+
+    HashMap<String, Integer> getDaysMap() {
+        return daysMap;
+    }
+
+    public int getSummaryCount() {
+        return matchupSum;
+    }
+
+    public void add(int timeStamp) {
+        utcCalendar.setTimeInMillis(timeStamp * 1000L);
+        final Date time = utcCalendar.getTime();
+        final String dayString = TimeUtils.format(time, "yyyy-MM-dd");
+
+        Integer count = daysMap.get(dayString);
+        if (count == null) {
+            count = 1;
+            daysMap.put(dayString, count);
+        } else {
+            ++count;
+            daysMap.replace(dayString, count);
+        }
+
+        matchupSum++;
+    }
+
+    int getFileCount() {
+        return fileCount;
+    }
+
+    void countFile() {
+        ++fileCount;
+    }
+}

core/src/main/java/com/bc/fiduceo/qc/MmdQCTool.java

@@ -0,0 +1,148 @@
+package com.bc.fiduceo.qc;
+
+import com.bc.fiduceo.FiduceoConstants;
+import com.bc.fiduceo.log.FiduceoLogger;
+import com.bc.fiduceo.util.NetCDFUtils;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import ucar.ma2.Array;
+import ucar.ma2.IndexIterator;
+import ucar.ma2.InvalidRangeException;
+import ucar.nc2.NetcdfFile;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.*;
+import java.util.logging.Logger;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static com.bc.fiduceo.FiduceoConstants.VERSION_NUMBER;
+import static com.bc.fiduceo.util.MMDUtil.getMMDFileNamePattern;
+
+class MmdQCTool {
+
+    private final static Logger logger = FiduceoLogger.getLogger();
+
+    private FileMessages fileMessages;
+    private MatchupAccumulator matchupAccumulator;
+
+    // package access for testing only tb 2023-02-14
+    static Options getOptions() {
+        final Options options = new Options();
+
+        final Option helpOption = new Option("h", "help", false, "Prints the tool usage.");
+        options.addOption(helpOption);
+
+        final Option inputOption = new Option("i", "input", true, "Defines the MMD input directory.");
+        inputOption.setRequired(true);
+        options.addOption(inputOption);
+
+        final Option timeOption = new Option("t", "time", true, "Defines matchup time variable name.");
+        timeOption.setRequired(true);
+        options.addOption(timeOption);
+
+        return options;
+    }
+
+    // package access for testing only tb 2023-02-14
+    static void writeReport(OutputStream outputStream, MatchupAccumulator accumulator, FileMessages fileMessages) {
+        final PrintWriter writer = new PrintWriter(outputStream);
+
+        final int fileCount = accumulator.getFileCount();
+        writer.println("Analysed " + fileCount + " file(s)");
+        if (fileCount == 0) {
+            writer.flush();
+            return;
+        }
+
+        writer.println();
+        final HashMap<String, List<String>> messageMap = fileMessages.getMessageMap();
+        final int size = messageMap.size();
+        writer.println(size + " file(s) with errors");
+        if (size > 0) {
+            // @todo 1 tb/tb add error messages per file here
+        }
+
+        writer.println();
+        writer.println("Total number of matchups: " + accumulator.getSummaryCount());
+        writer.println("Daily distribution:");
+
+        final TreeMap<String, Integer> treeMap = new TreeMap<>(accumulator.getDaysMap());
+        final Set<Map.Entry<String, Integer>> entries = treeMap.entrySet();
+        for (final Map.Entry<String, Integer> entry : entries) {
+            writer.println(entry.getKey() + ": " + entry.getValue());
+        }
+
+        writer.flush();
+    }
+
+    void run(CommandLine commandLine) throws IOException {
+        final String inputDirOption = commandLine.getOptionValue("i");
+        final List<Path> mmdFiles = getInputFiles(inputDirOption);
+        logger.info("Found " + mmdFiles.size() + " input file(s) to analyze.");
+
+        fileMessages = new FileMessages();
+        matchupAccumulator = new MatchupAccumulator();
+
+        // loop over files
+        for (final Path mmdFile : mmdFiles) {
+
+            try (final NetcdfFile netcdfFile = NetCDFUtils.openReadOnly(mmdFile.toAbsolutePath().toString())) {
+
+                // read time variable center pixel
+                final String timeVariableName = commandLine.getOptionValue("t");
+                final Array timeArray = NetCDFUtils.getCenterPosArrayFromMMDFile(netcdfFile, timeVariableName, null,
+                        null, FiduceoConstants.MATCHUP_COUNT);
+                final IndexIterator iterator = timeArray.getIndexIterator();
+                while (iterator.hasNext()) {
+                    final int time = iterator.getIntNext();
+                    matchupAccumulator.add(time);
+                }
+            } catch (IOException | InvalidRangeException ioException) {
+                fileMessages.add(mmdFile.getFileName().toString(), ioException.getMessage());
+            }
+        }
+
+        // write report
+        final TreeMap<String, Integer> treeMap = new TreeMap<>(matchupAccumulator.getDaysMap());
+        final Set<Map.Entry<String, Integer>> entries = treeMap.entrySet();
+        for (final Map.Entry<String, Integer> entry : entries) {
+            System.out.println(entry.getKey() + ": " + entry.getValue());
+        }
+    }
+
+    private List<Path> getInputFiles(String inputDirOption) throws IOException {
+        final File inputDir = new File(inputDirOption);
+        if (!inputDir.isDirectory()) {
+            throw new IOException("Not a valid directory: " + inputDir.getAbsolutePath());
+        }
+
+        final Pattern pattern = getMMDFileNamePattern();
+        try (Stream<Path> pathStream = Files.walk(inputDir.toPath())) {
+            final Stream<Path> regularFiles = pathStream.filter(Files::isRegularFile);
+            final Stream<Path> mmdFileStream = regularFiles.filter(path -> pattern.matcher(path.getFileName().toString()).matches());
+            return mmdFileStream.collect(Collectors.toList());
+        }
+    }
+
+    static void printUsageTo(OutputStream outputStream) {
+        final String ls = System.lineSeparator();
+        final PrintWriter writer = new PrintWriter(outputStream);
+        writer.write("mmd-qc-tool version " + VERSION_NUMBER);
+        writer.write(ls + ls);
+
+        final HelpFormatter helpFormatter = new HelpFormatter();
+        helpFormatter.printHelp(writer, 120, "matchup-tool <options>", "Valid options are:",
+                getOptions(), 3, 3, "");
+
+        writer.flush();
+    }
+}

core/src/main/java/com/bc/fiduceo/qc/MmdQCToolMain.java

@@ -0,0 +1,34 @@
+package com.bc.fiduceo.qc;
+
+import com.bc.fiduceo.log.FiduceoLogger;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.PosixParser;
+
+public class MmdQCToolMain {
+
+    public static void main(String[] args) throws ParseException {
+        final MmdQCTool mmdQCTool = new MmdQCTool();
+
+        if (args.length == 0) {
+            MmdQCTool.printUsageTo(System.err);
+            return;
+        }
+
+        final CommandLineParser parser = new PosixParser();
+        final CommandLine commandLine = parser.parse(MmdQCTool.getOptions(), args);
+        if (commandLine.hasOption("h") || commandLine.hasOption("--help")) {
+            MmdQCTool.printUsageTo(System.err);
+            return;
+        }
+
+        try {
+            mmdQCTool.run(commandLine);
+        } catch (Throwable e) {
+            FiduceoLogger.getLogger().severe(e.getMessage());
+            e.printStackTrace();
+            System.exit(-1);
+        }
+    }
+}

core/src/main/java/com/bc/fiduceo/reader/ArrayCache.java

@@ -48,6 +48,12 @@ public ArrayCache(NetcdfFile netcdfFile) {
         variableFinder = netcdfFile::findVariable;
     }
 
+    public void clear() {
+        cache.clear();
+        scaledCache.clear();
+        injectedVariables.clear();
+    }
+
     // package access for testing only tb 2016-04-14
     static String createGroupedName(String groupName, String variableName) {
         return groupName + "_" + variableName;

core/src/main/java/com/bc/fiduceo/reader/insitu/sic_cci/AMSR2SectionParser.java

@@ -1,5 +1,6 @@
 package com.bc.fiduceo.reader.insitu.sic_cci;
 
+import com.bc.fiduceo.reader.netcdf.StringVariable;
 import com.bc.fiduceo.util.NetCDFUtils;
 import com.bc.fiduceo.util.VariableProxy;
 import ucar.ma2.DataType;
@@ -52,7 +53,7 @@ List<Variable> getVariables() {
         variables.add(new VariableProxy("AMSR2_scanpos", DataType.SHORT, attributes));
 
         attributes = new ArrayList<>();
-        variables.add(new VariableProxy("AMSR2_upstreamfile", DataType.CHAR, attributes));
+        variables.add(new StringVariable(new VariableProxy("AMSR2_upstreamfile", DataType.STRING, attributes), 48));
 
         attributes = new ArrayList<>();
         attributes.add(new Attribute(CF_FILL_VALUE_NAME, NetCDFUtils.getDefaultFillValue(short.class)));