Add a value type to the JSON decoder so whole numbers no longer change its sample type

README.md

@@ -132,11 +132,12 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul
    - **Purpose**: To parse JSON string data to extract a value and a timestamp, and to send data and domain samples based on this data.
    - **Main properties**:
       - *ValueKey* (string) — Specifies the JSON field name (or dot-separated path for nested objects) from which value data will be extracted. Use `.` to access nested fields, e.g. `"data.temperature"` extracts the `temperature` field from inside the `data` object. This property is required. It should be contained in the incoming JSON messages. Otherwise, a parsing error will occur.
+      - *ValueType* (list) — Defines the sample type of numeric values. By default it is set to *Float64* (1), which decodes every number as `Float64`, so the signal keeps its sample type when the values switch between whole numbers and decimals. If set to *Int64* (2), every number has to be a whole number, and a message with a fractional value fails to parse. If set to *Auto* (0), whole numbers are decoded as `Int64` and decimals as `Float64`, so the sample type of the signal changes whenever the type of the values does. String values are not affected.
       - *DomainMode* (list) —  Defines how the timestamp of the decoded signal is generated. By default it is set to *None* (0), which means that the decoded signal doesn't have a timestamp. If set to *Extract from message* (1), the JSON decoder will try to extract the timestamp from the incoming JSON messages (see *DomainKey* property). If set to *System time* (2), the timestamp of the decoded signal is set to the system time when the JSON message is received.
       - *DomainKey* (string) — Specifies the JSON field name (or dot-separated path for nested objects) from which the timestamp will be extracted. Dot notation is supported, e.g. `"info.timestamp"` extracts `timestamp` from inside the `info` object. This property is optional. If it is set it should be contained in the incoming JSON messages. Otherwise, a parsing error will occur.
       - *Unit* (string) — Specifies the unit symbol for the decoded value. This property is optional.
 
-   - **Supported value types**: the field addressed by *ValueKey* may hold a single value (integer, floating-point number or string) or an array of values. An array produces several samples at once, so in the *Extract from message* domain mode the *DomainKey* field has to be an array of the same size. The sample type of the output signal follows the type found in the message, and it is updated if the type changes. An array which mixes integers and floating-point numbers is decoded as a floating-point (`Float64`) array; an array of integers only is decoded as an `Int64` array. Any other mix of types within one array (for example numbers and strings) causes a parsing error.
+   - **Supported value types**: the field addressed by *ValueKey* may hold a single value (integer, floating-point number or string) or an array of values. An array produces several samples at once, so in the *Extract from message* domain mode the *DomainKey* field has to be an array of the same size. The sample type of numbers is set by *ValueType*. With *Auto*, the sample type of the output signal follows the type found in the message, and it is updated if the type changes: an array which mixes integers and floating-point numbers is decoded as a floating-point (`Float64`) array, and an array of integers only is decoded as an `Int64` array. Any other mix of types within one array (for example numbers and strings) causes a parsing error.
 
       Dot-notation paths support arbitrary nesting depth. For example, `"sensor.values.temperature"` traverses `sensor` → `values` → `temperature`.
       An element of an array is addressed by its index in square brackets: `"sensors[1].temperature"` takes the second element of the `sensors` array, and the indexes may be chained, e.g. `"matrix[1][0]"`. An index which is out of range, or an index applied to a field which is not an array, causes a parsing error.

modules/mqtt_streaming_module/include/mqtt_streaming_module/constants.h

@@ -39,6 +39,7 @@ static constexpr const char* PROPERTY_NAME_SUB_PREVIEW_SIGNAL_IS_STRING = "Messa
 static constexpr const char* PROPERTY_NAME_SUB_DATA_TIMEOUT = "DataTimeout";
 
 static constexpr const char* PROPERTY_NAME_DEC_VALUE_NAME = "ValueKey";
+static constexpr const char* PROPERTY_NAME_DEC_VALUE_TYPE = "ValueType";
 static constexpr const char* PROPERTY_NAME_DEC_TS_MODE = "DomainMode";
 static constexpr const char* PROPERTY_NAME_DEC_TS_NAME = "DomainKey";
 static constexpr const char* PROPERTY_NAME_DEC_UNIT = "Unit";

modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_json_decoder_fb_impl.h

@@ -44,6 +44,7 @@ class MqttJsonDecoderFbImpl final : public FunctionBlock
 
     struct FbConfig {
         std::string valueFieldName;
+        mqtt::MqttDataWrapper::ValueType valueType;
         mqtt::MqttDataWrapper::DomainSignalMode tsMode;
         std::string tsFieldName;
         std::string unitSymbol;

modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp

@@ -28,6 +28,7 @@ MqttJsonDecoderFbImpl::MqttJsonDecoderFbImpl(const ContextPtr& ctx,
 FunctionBlockTypePtr MqttJsonDecoderFbImpl::CreateType()
 {
     using DSM = mqtt::MqttDataWrapper::DomainSignalMode;
+    using DVT = mqtt::MqttDataWrapper::ValueType;
     auto defaultConfig = PropertyObject();
     {
         auto builder =
@@ -40,6 +41,21 @@ FunctionBlockTypePtr MqttJsonDecoderFbImpl::CreateType()
         defaultConfig.addProperty(builder.build());
     }
 
+    {
+        auto builder =
+            SelectionPropertyBuilder(PROPERTY_NAME_DEC_VALUE_TYPE,
+                                     List<IString>("Auto", "Float64", "Int64"),
+                                     static_cast<int>(DVT::Float64))
+                .setDescription(
+                    "Defines the sample type of numeric values. By default it is set to Float64, which decodes every number "
+                    "as a 64-bit floating-point value, so the signal keeps its sample type when the values switch between "
+                    "whole numbers and decimals. If set to Int64, every number has to be a whole number, and a message with "
+                    "a fractional value fails to parse. If set to Auto, whole numbers are decoded as Int64 and decimals as "
+                    "Float64, so the sample type of the signal changes whenever the type of the values does. String values "
+                    "are not affected.");
+        defaultConfig.addProperty(builder.build());
+    }
+
     {
         auto builder =
             SelectionPropertyBuilder(PROPERTY_NAME_DEC_TS_MODE,
@@ -113,6 +129,7 @@ void MqttJsonDecoderFbImpl::readProperties()
 {
     using namespace property_helper;
     using DSM = mqtt::MqttDataWrapper::DomainSignalMode;
+    using DVT = mqtt::MqttDataWrapper::ValueType;
     auto lock = this->getRecursiveConfigLock();
     configValid = true;
     configMsg.clear();
@@ -123,6 +140,18 @@ void MqttJsonDecoderFbImpl::readProperties()
         configValid = false;
     }
 
+    auto tmpValueType = readProperty<int, IInteger>(objPtr, PROPERTY_NAME_DEC_VALUE_TYPE, static_cast<int>(DVT::Float64));
+    if (tmpValueType < static_cast<int>(DVT::_count) && tmpValueType >= 0)
+    {
+        config.valueType = static_cast<DVT>(tmpValueType);
+    }
+    else
+    {
+        configMsg = fmt::format("Wrong value for the \"{}\" property!", PROPERTY_NAME_DEC_VALUE_TYPE);
+        configValid = false;
+        config.valueType = DVT::Float64;
+    }
+
     config.tsFieldName = readProperty<std::string, IString>(objPtr, PROPERTY_NAME_DEC_TS_NAME, "");
     config.unitSymbol = readProperty<std::string, IString>(objPtr, PROPERTY_NAME_DEC_UNIT, "");
     auto tmpTsMode = readProperty<int, IInteger>(objPtr, PROPERTY_NAME_DEC_TS_MODE, static_cast<int>(DSM::None));
@@ -151,6 +180,7 @@ void MqttJsonDecoderFbImpl::readProperties()
         config.tsFieldName = "";
     }
     jsonDataWorker.setValueFieldName(config.valueFieldName);
+    jsonDataWorker.setValueType(config.valueType);
     jsonDataWorker.setTimestampFieldName(config.tsFieldName);
     jsonDataWorker.setDomainSignalMode(config.tsMode);
     waitingData = configValid.load();

modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp

@@ -18,6 +18,7 @@
 using namespace daq;
 using namespace daq::modules::mqtt_streaming_module;
 using DDSM = mqtt::MqttDataWrapper::DomainSignalMode;
+using DVT = mqtt::MqttDataWrapper::ValueType;
 
 namespace
 {
@@ -424,6 +425,8 @@ class MqttJsonDecoderFbHelper : public DaqTestHelper
 {
 public:
     daq::FunctionBlockPtr decoderObj;
+    // The value type of the decoders this helper adds
+    DVT valueType = DVT::Float64;
 
     void onSignalsMessage(const mqtt::MqttMessage& msg)
     {
@@ -454,6 +457,7 @@ class MqttJsonDecoderFbHelper : public DaqTestHelper
 
         config.setPropertyValue(PROPERTY_NAME_DEC_TS_MODE, static_cast<int>(mode));
         config.setPropertyValue(PROPERTY_NAME_DEC_VALUE_NAME, valueF);
+        config.setPropertyValue(PROPERTY_NAME_DEC_VALUE_TYPE, static_cast<int>(valueType));
         config.setPropertyValue(PROPERTY_NAME_DEC_TS_NAME, tsF);
         config.setPropertyValue(PROPERTY_NAME_DEC_UNIT, unitSymbol);
         decoderObj = subMqttFb.addFunctionBlock(typeId, config);
@@ -495,6 +499,32 @@ class MqttJsonDecoderFbHelper : public DaqTestHelper
         return read<returnT>(reader, signal, 1000);
     }
 
+    // Sends ready-made JSON messages to a freshly created decoder FB and returns the data packets it sends
+    std::vector<DataPacketPtr> transferRawMessages(const std::vector<std::string>& jsons, const std::string& valueF)
+    {
+        const auto topic = buildTopicName();
+        CreateDecoderFB(topic, valueF, DDSM::None, "");
+
+        auto reader = daq::PacketReader(getSignals()[0]);
+        for (const auto& json : jsons)
+            onSignalsMessage({topic, std::vector<uint8_t>(json.begin(), json.end()), 1, 0});
+
+        std::vector<DataPacketPtr> packets;
+        auto timer = helper::utils::Timer(1000);
+        while (!reader.getEmpty() || !timer.expired())
+        {
+            if (reader.getEmpty())
+            {
+                std::this_thread::sleep_for(std::chrono::milliseconds(20));
+                continue;
+            }
+            const auto packet = reader.read();
+            if (packet.getType() == PacketType::Data)
+                packets.push_back(packet.asPtr<IDataPacket>());
+        }
+        return packets;
+    }
+
     template <typename vT, typename tsT>
     std::vector<std::pair<std::vector<vT>, std::vector<uint64_t>>>
     transferData(const std::vector<std::pair<std::vector<vT>, std::vector<tsT>>>& data, const std::string& jsonDataTemplate)
@@ -649,13 +679,18 @@ TEST_F(MqttJsonDecoderFbTest, DefaultConfig)
 
     ASSERT_TRUE(defaultConfig.assigned());
 
-    EXPECT_EQ(defaultConfig.getAllProperties().getCount(), 4u);
+    EXPECT_EQ(defaultConfig.getAllProperties().getCount(), 5u);
 
     ASSERT_TRUE(defaultConfig.hasProperty(PROPERTY_NAME_DEC_VALUE_NAME));
     ASSERT_EQ(defaultConfig.getProperty(PROPERTY_NAME_DEC_VALUE_NAME).getValueType(), CoreType::ctString);
     EXPECT_EQ(defaultConfig.getPropertyValue(PROPERTY_NAME_DEC_VALUE_NAME).asPtr<IString>().getLength(), 0u);
     EXPECT_TRUE(defaultConfig.getProperty(PROPERTY_NAME_DEC_VALUE_NAME).getVisible());
 
+    ASSERT_TRUE(defaultConfig.hasProperty(PROPERTY_NAME_DEC_VALUE_TYPE));
+    ASSERT_EQ(defaultConfig.getProperty(PROPERTY_NAME_DEC_VALUE_TYPE).getValueType(), CoreType::ctInt);
+    EXPECT_EQ(defaultConfig.getPropertyValue(PROPERTY_NAME_DEC_VALUE_TYPE).asPtr<IInteger>(), static_cast<int>(DVT::Float64));
+    EXPECT_TRUE(defaultConfig.getProperty(PROPERTY_NAME_DEC_VALUE_TYPE).getVisible());
+
     ASSERT_TRUE(defaultConfig.hasProperty(PROPERTY_NAME_DEC_TS_MODE));
     ASSERT_EQ(defaultConfig.getProperty(PROPERTY_NAME_DEC_TS_MODE).getValueType(), CoreType::ctInt);
     EXPECT_EQ(defaultConfig.getPropertyValue(PROPERTY_NAME_DEC_TS_MODE).asPtr<IInteger>(), static_cast<int>(DDSM::None));
@@ -877,6 +912,7 @@ TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalDoubleArrayWithoutDomain)
 
 TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalIntArray)
 {
+    valueType = DVT::Int64;
     std::vector<std::pair<std::vector<int64_t>, std::vector<uint64_t>>> dataToSend;
     merge(DATA_INT_INT_0, dataToSend);
     merge(DATA_INT_INT_1, dataToSend);
@@ -892,6 +928,7 @@ TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalIntArray)
 
 TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalIntArrayWithoutDomain)
 {
+    valueType = DVT::Int64;
     std::vector<std::pair<std::vector<int64_t>, std::vector<uint64_t>>> dataToSend;
     merge(DATA_INT_INT_0, dataToSend);
     merge(DATA_INT_INT_1, dataToSend);
@@ -940,7 +977,8 @@ TEST_F(MqttJsonDecoderFbTest, DataTransferMixedNumericArrayDoubleFirst)
 
 TEST_F(MqttJsonDecoderFbTest, DataTransferIntArrayIsNotPromoted)
 {
-    // An array of integers stays an integer array
+    // With the Auto value type an array of integers stays an integer array
+    valueType = DVT::Auto;
     const std::string json = R"json({"value": [1, -2, 3]})json";
     const std::vector<int64_t> expectedValues{1, -2, 3};
 
@@ -967,6 +1005,69 @@ TEST_F(MqttJsonDecoderFbTest, DataTransferMixedIncompatibleArray)
               std::string::npos);
 }
 
+TEST_F(MqttJsonDecoderFbTest, Float64ValueTypeKeepsTheSampleTypeAcrossWholeNumbers)
+{
+    // A publisher can write a whole number without a decimal point. The sample type must not follow it.
+    const auto packets = transferRawMessages({R"json({"value": 21})json", R"json({"value": 21.5})json", R"json({"value": 22})json"},
+                                             "value");
+    const std::vector<double> expectedValues{21.0, 21.5, 22.0};
+
+    ASSERT_EQ(packets.size(), expectedValues.size());
+    for (size_t i = 0; i < packets.size(); ++i)
+    {
+        ASSERT_EQ(packets[i].getDataDescriptor().getSampleType(), SampleType::Float64);
+        EXPECT_EQ(*static_cast<double*>(packets[i].getData()), expectedValues[i]);
+    }
+}
+
+TEST_F(MqttJsonDecoderFbTest, Float64ValueTypeConvertsAnIntegerArray)
+{
+    const auto packets = transferRawMessages({R"json({"value": [1, -2, 3]})json"}, "value");
+    const std::vector<double> expectedValues{1.0, -2.0, 3.0};
+
+    ASSERT_EQ(packets.size(), 1u);
+    ASSERT_EQ(packets[0].getDataDescriptor().getSampleType(), SampleType::Float64);
+    ASSERT_EQ(packets[0].getSampleCount(), expectedValues.size());
+    for (size_t i = 0; i < expectedValues.size(); ++i)
+        EXPECT_EQ(static_cast<double*>(packets[0].getData())[i], expectedValues[i]);
+}
+
+TEST_F(MqttJsonDecoderFbTest, AutoValueTypeFollowsEachMessage)
+{
+    valueType = DVT::Auto;
+    const auto packets = transferRawMessages({R"json({"value": 21})json", R"json({"value": 21.5})json"}, "value");
+
+    ASSERT_EQ(packets.size(), 2u);
+    EXPECT_EQ(packets[0].getDataDescriptor().getSampleType(), SampleType::Int64);
+    EXPECT_EQ(packets[1].getDataDescriptor().getSampleType(), SampleType::Float64);
+}
+
+TEST_F(MqttJsonDecoderFbTest, Int64ValueTypeAcceptsAWholeNumberWrittenAsADecimal)
+{
+    valueType = DVT::Int64;
+    const auto packets = transferRawMessages({R"json({"value": 21})json", R"json({"value": 22.0})json"}, "value");
+    const std::vector<int64_t> expectedValues{21, 22};
+
+    ASSERT_EQ(packets.size(), expectedValues.size());
+    for (size_t i = 0; i < packets.size(); ++i)
+    {
+        ASSERT_EQ(packets[i].getDataDescriptor().getSampleType(), SampleType::Int64);
+        EXPECT_EQ(*static_cast<int64_t*>(packets[i].getData()), expectedValues[i]);
+    }
+}
+
+TEST_F(MqttJsonDecoderFbTest, Int64ValueTypeRejectsAFraction)
+{
+    valueType = DVT::Int64;
+    const auto packets = transferRawMessages({R"json({"value": 21.5})json"}, "value");
+
+    EXPECT_EQ(packets.size(), 0u);
+    ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"),
+              Enumeration("ComponentStatusType", "Error", decoderObj.getContext().getTypeManager()));
+    EXPECT_NE(decoderObj.getStatusContainer().getStatusMessage("ComponentStatus").toStdString().find("not a whole number"),
+              std::string::npos);
+}
+
 TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalDoubleArrayDomainString)
 {
     std::vector<std::pair<std::vector<double>, std::vector<std::string>>> dataToSend;
@@ -984,6 +1085,7 @@ TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalDoubleArrayDomainString)
 
 TEST_P(MqttJsonFbIntDataPTest, DataTransferOneSignalInt)
 {
+    valueType = DVT::Int64;
     const auto dataToSend = GetParam();
     auto dataToReceive = transferData(dataToSend, VALID_JSON_DATA_0);
     ASSERT_EQ(dataToSend.size(), dataToReceive.size());
@@ -995,6 +1097,7 @@ TEST_P(MqttJsonFbIntDataPTest, DataTransferOneSignalInt)
 
 TEST_P(MqttJsonFbIntDataPTest, DataTransferOneSignalIntWithoutDomain)
 {
+    valueType = DVT::Int64;
     const auto dataToSend = GetParam();
     auto dataToReceive = transferDataWithoutDomain(dataToSend, VALID_JSON_DATA_1);
     ASSERT_EQ(dataToSend.size(), dataToReceive.size());

shared/mqtt_streaming_protocol/include/mqtt_streaming_protocol/MqttDataWrapper.h

@@ -47,13 +47,24 @@ class MqttDataWrapper final
         _count
     };
 
+    // Auto takes the sample type from each message, so a value that switches between a whole number and a
+    // decimal changes the signal's descriptor. Float64 and Int64 keep the sample type fixed.
+    enum class ValueType : int
+    {
+        Auto = 0,
+        Float64,
+        Int64,
+        _count
+    };
+
     MqttDataWrapper();
 
     void setOutputSignal(daq::SignalConfigPtr outputSignal);
     CmdResult createAndSendDataPacket(const std::string& json, const uint64_t externalTs);
     void setValueFieldName(std::string valueFieldName);
     void setTimestampFieldName(std::string tsFieldName);
     void setDomainSignalMode(DomainSignalMode mode);
+    void setValueType(ValueType type);
 
 private:
     struct DataPackets
@@ -79,6 +90,7 @@ class MqttDataWrapper final
     // used for description how to extract data from sample json
     MqttMsgDescriptor msgDescriptor;
     DomainSignalMode domainSignalMode;
+    ValueType valueType;
 
     std::pair<CmdResult, std::vector<DataPackets>> extractDataSamples(const std::string& json, const uint64_t externalTs);
     void sendDataSamples(const DataPackets& dataPackets);
@@ -107,6 +119,7 @@ class MqttDataWrapper final
     bool extractValue(ExtractionContext& ctx, const rapidjson::Value& node);
     bool extractTimestamp(ExtractionContext& ctx, const rapidjson::Value& node);
     bool validateExtractionResult(ExtractionContext& ctx);
+    bool convertValueType(ExtractionContext& ctx);
     bool buildPackets(ExtractionContext& ctx, const uint64_t externalTs);
 
     template <typename TReadType>