From 7c1f3ce10c83550060e85e1c98bcf4cd66b85b2f Mon Sep 17 00:00:00 2001 From: Martin Kraner Date: Tue, 13 Oct 2026 11:31:58 +0200 Subject: [PATCH] Add a value type to the JSON decoder so whole numbers no longer change its sample type --- README.md | 3 +- .../include/mqtt_streaming_module/constants.h | 1 + .../mqtt_json_decoder_fb_impl.h | 1 + .../src/mqtt_json_decoder_fb_impl.cpp | 30 +++++ .../tests/test_mqtt_json_decoder_fb.cpp | 107 +++++++++++++++++- .../mqtt_streaming_protocol/MqttDataWrapper.h | 13 +++ .../src/MqttDataWrapper.cpp | 56 ++++++++- 7 files changed, 207 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 8390a29e..2f3f6872 100644 --- a/README.md +++ b/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. diff --git a/modules/mqtt_streaming_module/include/mqtt_streaming_module/constants.h b/modules/mqtt_streaming_module/include/mqtt_streaming_module/constants.h index 67b5bb18..6089abd2 100644 --- a/modules/mqtt_streaming_module/include/mqtt_streaming_module/constants.h +++ b/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"; diff --git a/modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_json_decoder_fb_impl.h b/modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_json_decoder_fb_impl.h index 9f322111..c5974f2a 100644 --- a/modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_json_decoder_fb_impl.h +++ b/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; diff --git a/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp b/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp index cdfb9651..fcd510ec 100644 --- a/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp +++ b/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("Auto", "Float64", "Int64"), + static_cast(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(objPtr, PROPERTY_NAME_DEC_VALUE_TYPE, static_cast(DVT::Float64)); + if (tmpValueType < static_cast(DVT::_count) && tmpValueType >= 0) + { + config.valueType = static_cast(tmpValueType); + } + else + { + configMsg = fmt::format("Wrong value for the \"{}\" property!", PROPERTY_NAME_DEC_VALUE_TYPE); + configValid = false; + config.valueType = DVT::Float64; + } + config.tsFieldName = readProperty(objPtr, PROPERTY_NAME_DEC_TS_NAME, ""); config.unitSymbol = readProperty(objPtr, PROPERTY_NAME_DEC_UNIT, ""); auto tmpTsMode = readProperty(objPtr, PROPERTY_NAME_DEC_TS_MODE, static_cast(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(); diff --git a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp index b88abb2c..64b2266b 100644 --- a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp +++ b/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(mode)); config.setPropertyValue(PROPERTY_NAME_DEC_VALUE_NAME, valueF); + config.setPropertyValue(PROPERTY_NAME_DEC_VALUE_TYPE, static_cast(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(reader, signal, 1000); } + // Sends ready-made JSON messages to a freshly created decoder FB and returns the data packets it sends + std::vector transferRawMessages(const std::vector& 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(json.begin(), json.end()), 1, 0}); + + std::vector 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()); + } + return packets; + } + template std::vector, std::vector>> transferData(const std::vector, std::vector>>& 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().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(), static_cast(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(), static_cast(DDSM::None)); @@ -877,6 +912,7 @@ TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalDoubleArrayWithoutDomain) TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalIntArray) { + valueType = DVT::Int64; std::vector, std::vector>> 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::vector>> 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 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 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(packets[i].getData()), expectedValues[i]); + } +} + +TEST_F(MqttJsonDecoderFbTest, Float64ValueTypeConvertsAnIntegerArray) +{ + const auto packets = transferRawMessages({R"json({"value": [1, -2, 3]})json"}, "value"); + const std::vector 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(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 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(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::vector>> 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()); diff --git a/shared/mqtt_streaming_protocol/include/mqtt_streaming_protocol/MqttDataWrapper.h b/shared/mqtt_streaming_protocol/include/mqtt_streaming_protocol/MqttDataWrapper.h index ea7c3157..5806f6ed 100644 --- a/shared/mqtt_streaming_protocol/include/mqtt_streaming_protocol/MqttDataWrapper.h +++ b/shared/mqtt_streaming_protocol/include/mqtt_streaming_protocol/MqttDataWrapper.h @@ -47,6 +47,16 @@ 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); @@ -54,6 +64,7 @@ class MqttDataWrapper final 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> 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 diff --git a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp index 3f26ff32..37bcf452 100644 --- a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp +++ b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp @@ -6,6 +6,8 @@ #include #include #include +#include +#include #include #include @@ -284,7 +286,8 @@ namespace mqtt { MqttDataWrapper::MqttDataWrapper() - : domainSignalMode(DomainSignalMode::None) + : domainSignalMode(DomainSignalMode::None), + valueType(ValueType::Float64) { } @@ -321,6 +324,11 @@ void MqttDataWrapper::setDomainSignalMode(DomainSignalMode mode) domainSignalMode = mode; } +void MqttDataWrapper::setValueType(ValueType type) +{ + valueType = type; +} + bool MqttDataWrapper::parseJson(ExtractionContext& ctx, const std::string& json) { ctx.jsonParsed = true; @@ -525,6 +533,51 @@ bool MqttDataWrapper::validateExtractionResult(ExtractionContext& ctx) return (valueExtractionError || tsExtractionError || alignmentError); } +bool MqttDataWrapper::convertValueType(ExtractionContext& ctx) +{ + if (!ctx.result.success) + return false; + + if (valueType == ValueType::Float64) + { + if (const auto* integers = std::get_if>(&ctx.value)) + { + std::vector numbers; + numbers.reserve(integers->size()); + std::transform(integers->cbegin(), + integers->cend(), + std::back_inserter(numbers), + [](int64_t integer) { return static_cast(integer); }); + ctx.value = std::move(numbers); + } + } + else if (valueType == ValueType::Int64) + { + if (const auto* numbers = std::get_if>(&ctx.value)) + { + // 2^63, the first whole number past the end of the Int64 range + constexpr double int64Bound = 9223372036854775808.0; + + std::vector integers; + integers.reserve(numbers->size()); + for (const double number : *numbers) + { + if (std::trunc(number) != number || number < -int64Bound || number >= int64Bound) + { + ctx.result.addError( + fmt::format("The value type is Int64, but '{}' holds {}, which is not a whole number in the Int64 range. ", + msgDescriptor.valueFieldName, + number)); + return false; + } + integers.push_back(static_cast(number)); + } + ctx.value = std::move(integers); + } + } + return true; +} + bool MqttDataWrapper::buildPackets(ExtractionContext& ctx, const uint64_t externalTs) { ctx.dataPacketBuilt = false; @@ -556,6 +609,7 @@ std::pair> MqttDataWrapper: parseJson(ctx, json); parseJsonFields(ctx); validateExtractionResult(ctx); + convertValueType(ctx); buildPackets(ctx, externalTs); return std::pair{std::move(ctx.result), std::move(ctx.outputData)};