Lumped Forcings Engine Data Provider

data/forcing/forcings-engine/.gitignore

@@ -0,0 +1,8 @@
+gfs/**/*.grib2
+
+# Temp directory used by the forcings engine instances
+scratch/*
+!scratch/.keep
+
+# Cropped GRIB files
+!gfs/gfs.20240117/00/atmos/gfs.t00z.sfluxgrbf*.grib2

data/forcing/forcings-engine/config_aorc.yml

@@ -0,0 +1,64 @@
+time_step_seconds: 3600
+initial_time: 0
+NWM_VERSION: 4.0
+NWM_CONFIG: "AORC"
+InputForcings: [12] 
+InputForcingDirectories: ['']
+InputForcingTypes: ["NETCDF4"]
+InputMandatory: [1]
+OutputFrequency: 60
+SubOutputHour: 0
+SubOutFreq: 0
+ScratchDir: "data/forcing/forcings-engine/scratch"
+Output: 1
+compressOutput: 0
+floatOutput: 0
+AnAFlag: 0
+LookBack : -9999
+RefcstBDateProc: "202301170000"
+RefcstEDateProc: "202301170100"
+ForecastFrequency: 60
+ForecastShift: 0
+ForecastInputHorizons: [14400]
+ForecastInputOffsets: [0]
+GeogridIn: "data/forcing/forcings-engine/Gauge_01073000_ESMF_Mesh.nc"
+SpatialMetaIn: ""
+GRID_TYPE: "hydrofabric"
+NodeCoords: 'nodeCoords'
+ElemCoords: 'centerCoords'
+ElemConn: 'elementConn'
+ElemID: 'element_id'
+NumElemConn: 'numElementConn'
+HGTVAR: 'Element_Elevation'
+SLOPE: 'Element_Slope'
+SLOPE_AZIMUTH: 'Element_Slope_Azmuith'
+IgnoredBorderWidths: [0]
+RegridOpt: [1]
+ForcingTemporalInterpolation: [0]
+TemperatureBiasCorrection: [0]
+PressureBiasCorrection: [0]
+HumidityBiasCorrection: [0]
+WindBiasCorrection: [0]
+SwBiasCorrection: [0]
+LwBiasCorrection: [0]
+PrecipBiasCorrection: [0]
+TemperatureDownscaling: [0]
+ShortwaveDownscaling: [0]
+PressureDownscaling: [0]
+PrecipDownscaling: [0]
+HumidityDownscaling: [0]
+DownscalingParamDirs: ["data/forcing/forcings-engine/scratch"]
+SuppPcp: []
+SuppPcpForcingTypes: ''
+SuppPcpDirectories: ''
+SuppPcpParamDir: ''
+RegridOptSuppPcp: []
+SuppPcpTemporalInterpolation: []
+SuppPcpInputOffsets: []
+SuppPcpMandatory: []
+RqiMethod: 0
+RqiThreshold: 0.9
+cfsEnsNumber: ''
+custom_input_fcst_freq: []
+includeLQFrac: 0
+

include/forcing/DataProvider.hpp

@@ -21,14 +21,17 @@ namespace data_access
             BACK_FILL
     };
 
-    template <class data_type, class selection_type> class DataProvider
+    template <class DataType, class SelectionType> class DataProvider
     {
         /** This class provides a generic interface to data services
         *
         */
 
         public:
 
+        using data_type = DataType;
+        using selection_type = SelectionType;
+
         virtual ~DataProvider() = default;
 
         /**

include/forcing/ForcingsEngineDataProvider.hpp

@@ -33,7 +33,7 @@ namespace detail {
 
 //! Parse time string from format.
 //! Utility function for ForcingsEngineLumpedDataProvider constructor.
-time_t parse_time(const std::string& time, const std::string& fmt);
+time_t parse_time(const std::string& time, const std::string& fmt = default_time_format);
 
 //! Check that requirements for running the forcings engine
 //! are available at runtime. If requirements are not available,

include/forcing/ForcingsEngineLumpedDataProvider.hpp

@@ -0,0 +1,49 @@
+#pragma once
+
+#include <NGenConfig.h>
+
+#if NGEN_WITH_PYTHON
+
+#include <forcing/ForcingsEngineDataProvider.hpp>
+#include <forcing/DataProviderSelectors.hpp>
+
+namespace data_access {
+
+struct ForcingsEngineLumpedDataProvider final :
+  public ForcingsEngineDataProvider<double, CatchmentAggrDataSelector>
+{
+    using base_type = ForcingsEngineDataProvider<data_type, selection_type>;
+
+    ~ForcingsEngineLumpedDataProvider() override = default;
+
+    ForcingsEngineLumpedDataProvider(
+        const std::string& init,
+        std::size_t time_begin_seconds,
+        std::size_t time_end_seconds,
+        const std::string& divide_id
+    );
+
+    data_type get_value(
+        const selection_type& selector,
+        data_access::ReSampleMethod m
+    ) override;
+
+    std::vector<data_type> get_values(
+        const selection_type& selector,
+        data_access::ReSampleMethod m
+    ) override;
+
+    //! Get this provider's Divide ID.
+    std::size_t divide() const noexcept;
+
+    //! Get this provider's Divide ID index within the Forcings Engine.
+    std::size_t divide_index() const noexcept;
+
+  private:
+    std::size_t divide_id_;
+    std::size_t divide_idx_;
+};
+
+} // namespace data_access
+
+#endif // NGEN_WITH_PYTHON

src/forcing/CMakeLists.txt

@@ -31,6 +31,7 @@ if(NGEN_WITH_PYTHON)
     target_sources(forcing
       PRIVATE
         "${CMAKE_CURRENT_LIST_DIR}/ForcingsEngineDataProvider.cpp"
+        "${CMAKE_CURRENT_LIST_DIR}/ForcingsEngineLumpedDataProvider.cpp"
     )
     target_link_libraries(forcing PUBLIC pybind11::embed NGen::ngen_bmi)
 endif()

src/forcing/ForcingsEngineLumpedDataProvider.cpp

@@ -0,0 +1,134 @@
+#include "DataProvider.hpp"
+#include <chrono>
+#include <forcing/ForcingsEngineLumpedDataProvider.hpp>
+
+namespace data_access {
+
+using Provider     = ForcingsEngineLumpedDataProvider;
+using BaseProvider = Provider::base_type;
+
+std::size_t convert_divide_id_stoi(const std::string& divide_id)
+{
+    auto separator = divide_id.find('-');
+    const char* split = (
+        separator == std::string::npos
+        ? &divide_id[0]
+        : &divide_id[separator + 1]
+    );
+
+    return std::atol(split);
+}
+
+Provider::ForcingsEngineLumpedDataProvider(
+    const std::string& init,
+    std::size_t time_begin_seconds,
+    std::size_t time_end_seconds,
+    const std::string& divide_id
+)
+  : BaseProvider(init, time_begin_seconds, time_end_seconds)
+{
+    divide_id_ = convert_divide_id_stoi(divide_id);
+
+    // Check that CAT-ID is an available output name, otherwise we most likely aren't
+    // running the correct configuration of the forcings engine for this class.
+    const auto cat_id_pos = std::find(var_output_names_.begin(), var_output_names_.end(), "CAT-ID");
+    if (cat_id_pos == var_output_names_.end()) {
+        throw std::runtime_error{
+            "Failed to initialize ForcingsEngineLumpedDataProvider: `CAT-ID` is not an output variable of the forcings engine."
+            " Does " + init + " have `GRID_TYPE` set to 'hydrofabric'?"
+        };
+    }
+    var_output_names_.erase(cat_id_pos);
+
+    const auto size_id_dimension = static_cast<std::size_t>(
+        bmi_->GetVarNbytes("CAT-ID") / bmi_->GetVarItemsize("CAT-ID")
+    );
+
+    // Copy CAT-ID values into instance vector
+    const auto cat_id_span = boost::span<const int>(
+        static_cast<const int*>(bmi_->GetValuePtr("CAT-ID")),
+        size_id_dimension
+    );
+
+    auto divide_id_pos = std::find(cat_id_span.begin(), cat_id_span.end(), divide_id_);
+    if (divide_id_pos == cat_id_span.end()) {
+        // throw std::runtime_error{"Unable to find divide ID `" + divide_id + "` in given Forcings Engine domain"};
+        divide_idx_ = static_cast<std::size_t>(-1);
+    } else {
+        divide_idx_ = std::distance(cat_id_span.begin(), divide_id_pos);
+    }
+}
+
+std::size_t Provider::divide() const noexcept
+{
+    return divide_id_;
+}
+
+std::size_t Provider::divide_index() const noexcept
+{
+    return divide_idx_;
+}
+
+Provider::data_type Provider::get_value(
+    const Provider::selection_type& selector,
+    data_access::ReSampleMethod m
+)
+{
+    assert(divide_id_ == convert_divide_id_stoi(selector.get_id()));
+
+    auto variable = ensure_variable(selector.get_variable_name());
+
+    if (m == ReSampleMethod::SUM || m == ReSampleMethod::MEAN) {
+        double acc = 0.0;
+        const auto start = clock_type::from_time_t(selector.get_init_time());
+        assert(start >= time_begin_);
+
+        const auto end = std::chrono::seconds{selector.get_duration_secs()} + start;
+        assert(end <= time_end_);
+
+        auto current = start;
+        while (current < end) {
+            current += time_step_;
+            bmi_->UpdateUntil(std::chrono::duration_cast<std::chrono::seconds>(current - time_begin_).count());
+            acc += static_cast<double*>(bmi_->GetValuePtr(variable))[divide_idx_];
+        }
+
+        if (m == ReSampleMethod::MEAN) {
+            auto duration = std::chrono::duration_cast<std::chrono::seconds>(current - start).count();
+            auto num_time_steps = duration / time_step_.count();
+            acc /= num_time_steps;
+        }
+
+        return acc;
+    }
+
+    throw std::runtime_error{"Given ReSampleMethod " + std::to_string(m) + " not implemented."};
+}
+
+std::vector<Provider::data_type> Provider::get_values(
+    const Provider::selection_type& selector,
+    data_access::ReSampleMethod /* unused */
+)
+{
+    assert(divide_id_ == convert_divide_id_stoi(selector.get_id()));
+
+    auto variable = ensure_variable(selector.get_variable_name());
+
+    const auto start = clock_type::from_time_t(selector.get_init_time());
+    assert(start >= time_begin_);
+
+    const auto end = std::chrono::seconds{selector.get_duration_secs()} + start;
+    assert(end <= time_end_);
+
+    std::vector<double> values;
+    auto current = start;
+    while (current < end) {
+        current += time_step_;
+        bmi_->UpdateUntil(std::chrono::duration_cast<std::chrono::seconds>(current - time_begin_).count());
+        values.push_back(static_cast<double*>(bmi_->GetValuePtr(variable))[divide_idx_]);
+    }
+
+    return values;
+}
+
+} // namespace data_access

test/CMakeLists.txt

@@ -187,6 +187,7 @@ ngen_add_test(
     test_forcings_engine
     OBJECTS
         forcing/ForcingsEngineDataProvider_Test.cpp
+        forcing/ForcingsEngineLumpedDataProvider_Test.cpp
     LIBRARIES
         NGen::forcing
     REQUIRES