Merge pull request #38 from ecmwf/feature/scaling-action

Feature/scaling action

src/multio/action/CMakeLists.txt

@@ -13,3 +13,4 @@ add_subdirectory(null)
 add_subdirectory(interpolate)
 add_subdirectory(interpolate-fesom)
 add_subdirectory(select)
+add_subdirectory(scale)

src/multio/action/scale/CMakeLists.txt

@@ -0,0 +1,24 @@
+ecbuild_add_library(
+
+    TARGET multio-action-scale
+
+    TYPE SHARED # Due to reliance on factory self registration this library cannot be static
+
+    SOURCES
+        Scale.cc
+        Scale.h
+        Mapping.cc
+        Mapping.h 
+        Scaling.cc
+        Scaling.h
+        MetadataUtils.cc
+        MetadataUtils.h
+
+    PRIVATE_INCLUDES
+        ${ECKIT_INCLUDE_DIRS}
+
+    CONDITION
+
+    PUBLIC_LIBS
+        multio
+)

src/multio/action/scale/Mapping.cc

@@ -0,0 +1,44 @@
+#include "Mapping.h"
+
+#include <cstdlib>
+#include <string>
+
+#include "multio/LibMultio.h"
+#include "multio/message/Glossary.h"
+#include "multio/util/Substitution.h"
+#include "MetadataUtils.h"
+
+
+namespace multio::action {
+
+using message::glossary;
+
+ScaleMapping::ScaleMapping(const config::ComponentConfiguration& compConf) : hasMapping_(false), scaleMap_{} {
+
+    const auto mappings = compConf.parsedConfig().has("mapping-definition")
+                            ? compConf.parsedConfig().getSubConfigurations("mapping-definition")
+                            : std::vector<eckit::LocalConfiguration>{};
+
+    if (!mappings.empty()) {
+        hasMapping_ = true;
+        for (const auto& mapping : mappings) {
+            auto matcher = mapping.getSubConfiguration("case");
+            scaleMap_[matcher.getString("param-is")] = matcher.getString("map-to-param");
+        }
+    }
+}
+
+void ScaleMapping::applyMapping(message::Metadata& md) const {
+    if (hasMapping_) {
+
+        std::string cparam = extractParam(md);
+
+        auto it = scaleMap_.find(cparam);
+        if (it != scaleMap_.end()) {
+            md.set(glossary().paramId, std::stoll(it->second)); 
+            md.set(glossary().param, it->second.c_str());
+        }
+    }
+}
+
+}  // namespace multio::action

src/multio/action/scale/Mapping.h

@@ -0,0 +1,22 @@
+#pragma once
+
+#include <map>
+#include <string>
+
+#include "eckit/config/LocalConfiguration.h"
+#include "multio/config/ComponentConfiguration.h"
+#include "multio/message/Message.h"
+
+namespace multio::action {
+
+class ScaleMapping {
+private:
+    bool hasMapping_;
+    std::map<std::string, std::string> scaleMap_;
+
+public:
+    explicit ScaleMapping(const config::ComponentConfiguration& compConf);
+    void applyMapping(message::Metadata& md) const;
+};
+
+}  // namespace multio::action

src/multio/action/scale/MetadataUtils.cc

@@ -0,0 +1,26 @@
+
+#include "MetadataUtils.h"
+#include "eckit/exception/Exceptions.h"
+#include "multio/message/Metadata.h"
+
+
+#include "multio/message/Glossary.h"
+
+
+namespace multio::action {
+using message::glossary;
+
+std::string extractParam(const multio::message::Metadata& md) {
+    std::string cparam{};
+    if (auto param = md.getOpt<std::string>(glossary().param); param) {
+        cparam = *param;
+    } else if (auto paramId = md.getOpt<std::int64_t>(glossary().paramId); paramId) {
+        cparam = std::to_string(*paramId);
+    }
+    else {
+        throw eckit::SeriousBug{"param/paramId metadata not present", Here()};
+    }
+
+    return cparam;
+}
+}

src/multio/action/scale/MetadataUtils.h

@@ -0,0 +1,11 @@
+#pragma once
+
+#include <string>
+#include "multio/LibMultio.h"
+#include "multio/action/scale/MetadataUtils.h"
+#include "multio/message/Metadata.h"
+
+namespace multio::action {
+    std::string extractParam(const multio::message::Metadata& md);
+
+}

src/multio/action/scale/Scale.cc

@@ -0,0 +1,96 @@
+/*
+ * (C) Copyright 1996- ECMWF.
+ *
+ * This software is licensed under the terms of the Apache Licence Version 2.0
+ * which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
+ * In applying this licence, ECMWF does not waive the privileges and immunities
+ * granted to it by virtue of its status as an intergovernmental organisation nor
+ * does it submit to any jurisdiction.
+ */
+
+#include "multio/action/scale/Scale.h"
+
+#include "eckit/exception/Exceptions.h"
+#include "eckit/log/Log.h"
+
+
+#include "multio/LibMultio.h"
+#include "multio/config/ComponentConfiguration.h"
+#include "multio/message/Glossary.h"
+#include "multio/message/Message.h"
+#include "multio/util/PrecisionTag.h"
+
+#include "multio/action/scale/Mapping.h"
+#include "multio/action/scale/Scaling.h"
+#include "multio/action/scale/MetadataUtils.h"
+
+
+namespace multio::action {
+
+using message::glossary;
+
+Scale::Scale(const ComponentConfiguration& compConf) :
+    ChainedAction(compConf), scaling_{compConf}, mapping_{compConf}, paramsToScale_{} {
+
+    const auto mappings = compConf.parsedConfig().has("mapping-definition")
+                            ? compConf.parsedConfig().getSubConfigurations("mapping-definition")
+                            : throw eckit::SeriousBug{"Scaling information not specified in plan", Here()};
+
+    if (!mappings.empty()) {
+        for (const auto& mapping : mappings) {
+            auto matcher = mapping.getSubConfiguration("case");
+            paramsToScale_.insert(matcher.getString("param-is"));
+        }
+    }
+}
+
+void Scale::executeImpl(message::Message msg) {
+    if (msg.tag() != message::Message::Tag::Field) {
+        executeNext(std::move(msg));
+        return;
+    }
+
+    std::string cparam = extractParam(msg.metadata());
+
+    // Continue if no scaling definition was specified in the plan.
+    if (paramsToScale_.find(cparam) == paramsToScale_.end()) {
+        executeNext(std::move(msg));
+        return;
+    }
+    //Scale the message
+    util::dispatchPrecisionTag(msg.precision(), [&](auto pt) {
+        using Precision = typename decltype(pt)::type;
+        ScaleMessage<Precision>(msg); // Modify msg in place
+    });
+    //pass on the modified message
+    executeNext(std::move(msg));
+    return;
+}
+
+template <typename Precision>
+void Scale::ScaleMessage(message::Message& msg) const {
+
+    LOG_DEBUG_LIB(LibMultio) << "Scale :: Metadata of the input message :: Apply Scaling " << std::endl
+                             << msg.metadata() << std::endl;
+    msg.acquire();
+
+    // Potentially work with msg = scaling_.applyScaling<Precision>(std::move(msg))
+    scaling_.applyScaling<Precision>(msg);
+
+    mapping_.applyMapping(msg.modifyMetadata());
+
+    return;
+}
+
+void Scale::print(std::ostream& os) const {
+    os << "Scale Action ";
+}
+
+static ActionBuilder<Scale> ScaleBuilder("scale");
+
+
+template void Scale::ScaleMessage<float>(message::Message&) const;
+template void Scale::ScaleMessage<double>(message::Message&) const;
+
+
+}  // namespace multio::action

src/multio/action/scale/Scale.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include "multio/action/ChainedAction.h"
+#include "multio/config/ComponentConfiguration.h" 
+
+#include "multio/action/scale/Mapping.h"
+#include "multio/action/scale/Scaling.h"
+
+namespace multio::action {
+
+class Scale final : public ChainedAction {
+public:
+    explicit Scale(const ComponentConfiguration& compConf);  // Constructor declaration
+
+    void executeImpl(message::Message msg) override;
+
+private:
+    ScaleScaling scaling_;
+    ScaleMapping mapping_;
+    std::set<std::string> paramsToScale_;
+
+    template <typename Precision>
+    void ScaleMessage(message::Message& msg) const;
+
+    void print(std::ostream&) const override;
+};
+
+}  // namespace multio::action

src/multio/action/scale/Scaling.cc

@@ -0,0 +1,38 @@
+#include "Scaling.h"
+
+#include <cstdlib>
+#include <string>
+
+#include "multio/LibMultio.h"
+#include "multio/util/Substitution.h"
+
+
+namespace multio::action {
+
+
+ScaleScaling::ScaleScaling(const config::ComponentConfiguration& compConf) : hasScaling_(false), scaleFactor_{} {
+
+    const auto mappings = compConf.parsedConfig().has("mapping-definition")
+                            ? compConf.parsedConfig().getSubConfigurations("mapping-definition")
+                            : std::vector<eckit::LocalConfiguration>{};
+
+    if (!mappings.empty()) {
+        hasScaling_ = true;
+        for (const auto& mapping : mappings) {
+            auto matcher = mapping.getSubConfiguration("case");
+            scaleFactor_[matcher.getString("param-is")] = matcher.getDouble("scaling-factor");
+        }
+    }
+}
+double ScaleScaling::getScalingFactor(const std::string paramID) const {
+    auto it = scaleFactor_.find(paramID);
+    if (it != scaleFactor_.end()) {
+        return it->second;
+    }
+    else {
+        throw eckit::SeriousBug{"Scaling Factor not found paramID" + paramID, Here()};
+    }
+}
+
+
+}  // namespace multio::action

src/multio/action/scale/Scaling.h

@@ -0,0 +1,54 @@
+#pragma once
+
+#include <map>
+#include <string>
+
+#include "eckit/config/LocalConfiguration.h"
+#include "multio/config/ComponentConfiguration.h"
+#include "multio/message/Glossary.h"
+#include "multio/message/Message.h"
+#include "multio/action/scale/MetadataUtils.h"
+
+namespace multio::action {
+
+using message::glossary;
+
+class ScaleScaling {
+private:
+    bool hasScaling_;
+    std::map<std::string, double> scaleFactor_;
+
+public:
+    explicit ScaleScaling(const config::ComponentConfiguration& compConf);
+    double getScalingFactor(const std::string paramID) const;
+
+    template <typename Precision>
+    void applyScaling(message::Message& msg) const {
+        if (hasScaling_) {
+            std::string cparam = extractParam(msg.metadata());
+            // Find the scaling factor
+            double scaleFactor = getScalingFactor(cparam);
+
+            // Ensure the payload size is compatible with Precision type
+            size_t size = msg.payload().size() / sizeof(Precision);
+            if (size == 0) {
+                throw eckit::SeriousBug{" Payload is empty: Scaling Action: " + msg.metadata().toString(), Here()};
+            }
+
+            // Access the payload data
+            auto data = static_cast<Precision*>(msg.payload().modifyData());
+            if (!data) {
+                throw eckit::SeriousBug{
+                    " Payload data could not be modified: Scaling Action: " + msg.metadata().toString(), Here()};
+            }
+            // Apply the scaling factor using std::transform
+            std::transform(data, data + size, data,
+                           [scaleFactor](Precision value) { return static_cast<Precision>(value * scaleFactor); });
+        }
+        else {
+            return;
+        }
+    }
+};
+
+}  // namespace multio::action

src/multio/api/c/CMakeLists.txt

@@ -21,6 +21,7 @@ list( APPEND multio_action_plugins
     multio-action-transport
     multio-action-renumber-healpix
     multio-action-interpolate-fesom
+    multio-action-scale
 )
 
 if( HAVE_MIR )

src/multio/config/PlanConfiguration.cc

@@ -212,7 +212,6 @@ eckit::LocalConfiguration generate_interpolate_action(const metkit::mars::MarsPa
     return interpolate_action;
 };
 
-
 eckit::LocalConfiguration generate_encode_action(const metkit::mars::MarsParsedRequest& request,
                                                  const eckit::LocalConfiguration& componentConfig,
                                                  const std::string& templatesPath) {