diff --git a/core/opendaq/reader/include/opendaq/domain_value.h b/core/opendaq/reader/include/opendaq/domain_value.h new file mode 100644 index 0000000000..bffc941abc --- /dev/null +++ b/core/opendaq/reader/include/opendaq/domain_value.h @@ -0,0 +1,452 @@ +/* + * Copyright 2022-2025 openDAQ d.o.o. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include +#include +#include + +#include +#include +#include + +BEGIN_NAMESPACE_OPENDAQ + +struct DomainInfo +{ + std::chrono::system_clock::time_point epoch; + RatioPtr resolution; + + static DomainInfo fromDescriptor(const DataDescriptorPtr& descriptor) + { + if (!descriptor.assigned()) + DAQ_THROW_EXCEPTION(ArgumentNullException, "Descriptor must not be null"); + + auto epoch = daq::reader::parseEpoch(descriptor.getOrigin()); + auto resolution = descriptor.getTickResolution(); + + return {epoch, resolution}; + } + + friend bool operator==(const DomainInfo& lhs, const DomainInfo& rhs) + { + if (!lhs.resolution.assigned() || !rhs.resolution.assigned()) + DAQ_THROW_EXCEPTION(InvalidParameterException, "DomainInfo::resolution must be assigned."); + + if (!(lhs.epoch == rhs.epoch)) + return false; + if (!(lhs.resolution.getNumerator() == rhs.resolution.getNumerator())) + return false; + if (!(lhs.resolution.getDenominator() == rhs.resolution.getDenominator())) + return false; + return true; + } + + friend bool operator!=(const DomainInfo& lhs, const DomainInfo& rhs) + { + return !(lhs == rhs); + } +}; + +inline std::ostream& operator<<(std::ostream& os, const DomainInfo& info) +{ + os << "DomainInfo{" + << "epoch=" << info.epoch.time_since_epoch().count() << ", resolution=" << info.resolution.getNumerator() << "/" + << info.resolution.getDenominator() << "}"; + + return os; +} + +class DomainValue +{ +public: + explicit DomainValue(const DomainInfo& info) + : domain(info) + { + } + virtual ~DomainValue() = default; + + const DomainInfo& getDomain() const + { + return domain; + } + + virtual std::unique_ptr toCommonDomain(const DomainInfo& commonDomain) = 0; + virtual std::unique_ptr fromCommonDomain(const DomainInfo& regularDomain) = 0; + + virtual void roundUpOnDomainInterval(const RatioPtr& interval) = 0; + +#if !defined(NDEBUG) + virtual std::string asTime() const = 0; +#endif + + friend bool operator<(const DomainValue& lhs, const DomainValue& rhs) + { + return lhs.compare(rhs) < 0; + } + friend bool operator>(const DomainValue& lhs, const DomainValue& rhs) + { + return lhs.compare(rhs) > 0; + } + friend bool operator==(const DomainValue& lhs, const DomainValue& rhs) + { + return lhs.compare(rhs) == 0; + } + friend bool operator!=(const DomainValue& lhs, const DomainValue& rhs) + { + return lhs.compare(rhs) != 0; + } + +protected: + DomainInfo domain; + +private: + virtual int compare(const DomainValue& other) const = 0; +}; + +template +class DomainValueImpl : public DomainValue +{ +public: + explicit DomainValueImpl(const DomainInfo& info, Type value) + : DomainValue(info) + , value(value) + { +#if !defined(NDEBUG) + if constexpr (std::is_integral_v) + { + std::cout << "DomainValueImpl(value=" << value << ", domain=" << info << ")" << std::endl; + } +#endif + } + + ~DomainValueImpl() override = default; + + std::unique_ptr toCommonDomain(const DomainInfo& commonDomain) override + { + // Offset of current domain in common domain ticks + Int epochOffset = domain.epoch.time_since_epoch().count() - commonDomain.epoch.time_since_epoch().count(); + + using SysPeriod = std::chrono::system_clock::period; + Int scaleNumerator = SysPeriod::num * commonDomain.resolution.getDenominator(); + Int scaleDenominator = SysPeriod::den * commonDomain.resolution.getNumerator(); + Int offsetFromCommon = epochOffset * scaleNumerator / scaleDenominator; + + // tick_common = tick * multiplier + Int multiplierNumerator = domain.resolution.getNumerator() * commonDomain.resolution.getDenominator(); + Int multiplierDenominator = domain.resolution.getDenominator() * commonDomain.resolution.getNumerator(); + + Type valueScaledToCommon = 0; + if constexpr (std::is_integral_v) + { + // Round to the closest tick + valueScaledToCommon = static_cast((value / multiplierDenominator) * multiplierNumerator + + (2 * (value % multiplierDenominator) * multiplierNumerator + multiplierDenominator) / (2 * multiplierDenominator)); + } + else + { + valueScaledToCommon = static_cast(value * multiplierNumerator / static_cast(multiplierDenominator)); + } + Type valueInCommon = offsetFromCommon + valueScaledToCommon; + + return std::make_unique>(commonDomain, valueInCommon); + } + + std::unique_ptr fromCommonDomain(const DomainInfo& regularDomain) override + { + const auto& commonDomain = this->domain; + const auto& valueInCommon = this->value; + + // Offset of regularDomain domain in common domain ticks + Int epochOffset = regularDomain.epoch.time_since_epoch().count() - commonDomain.epoch.time_since_epoch().count(); + + using SysPeriod = std::chrono::system_clock::period; + Int scaleNumerator = SysPeriod::num * commonDomain.resolution.getDenominator(); + Int scaleDenominator = SysPeriod::den * commonDomain.resolution.getNumerator(); + Int offsetFromCommon = epochOffset * scaleNumerator / scaleDenominator; + Type valueScaledToCommon = valueInCommon - offsetFromCommon; + + Int multiplierNumerator = regularDomain.resolution.getNumerator() * commonDomain.resolution.getDenominator(); + Int multiplierDenominator = regularDomain.resolution.getDenominator() * commonDomain.resolution.getNumerator(); + + Type regularValue = 0; + if constexpr (std::is_integral_v) + { + // Round to the closest tick + regularValue = static_cast((valueScaledToCommon / multiplierNumerator) * multiplierDenominator + + (2 * (valueScaledToCommon % multiplierNumerator) * multiplierDenominator + multiplierNumerator) / (2 * multiplierNumerator)); + } + else + { + regularValue = static_cast(valueScaledToCommon * multiplierDenominator / static_cast(multiplierNumerator)); + } + + return std::make_unique>(regularDomain, regularValue); + } + + void roundUpOnDomainInterval(const RatioPtr& interval) override + { + auto num = domain.resolution.getNumerator() * interval.getDenominator(); + auto den = domain.resolution.getDenominator() * interval.getNumerator(); + + const Int gcd = std::gcd(num, den); + num /= gcd; + den /= gcd; + + if (den % num != 0) // 1 = k * num/den, the resolution is a fractional divider of a unit + DAQ_THROW_EXCEPTION(NotSupportedException, "Resolution must be aligned on full unit of domain"); + + value = static_cast((((value * num + den - 1) / den) * den) / num); + } + + Type getValue() const + { + return value; + } + +#if !defined(NDEBUG) + virtual std::string asTime() const override + { + using namespace reader; + + std::stringstream ss; + ss << toSysTime(value, domain.epoch, domain.resolution); + + return ss.str(); + } +#endif + + int compare(const DomainValue& other) const override + { + const auto* otherImpl = dynamic_cast*>(&other); + if (otherImpl == nullptr) + { + DAQ_THROW_EXCEPTION(InvalidParameterException, "Both DomainValue objects must be of the same type!"); + } + if (otherImpl->domain != this->domain) + DAQ_THROW_EXCEPTION(InvalidParameterException, "Have to compare DomainValue objects in the same domain!"); + + if (this->value > otherImpl->value) + return 1; + else if (this->value == otherImpl->value) + return 0; + else // this->value < otherImpl->value + return -1; + } + +private: + Type value; +}; + +template <> +class DomainValueImpl final : public DomainValue +{ +public: + using RangeValue = RangeType64::Type; + + explicit DomainValueImpl(const DomainInfo& info, RangeType64 value) + : DomainValue(info) + , value(value) + { + std::cout << "DomainValueImpl(value=" << value.start << ", domain=" << info << ")" << std::endl; + } + + std::unique_ptr toCommonDomain(const DomainInfo& commonDomain) override + { + // Offset of current domain in common domain ticks + Int epochOffset = domain.epoch.time_since_epoch().count() - commonDomain.epoch.time_since_epoch().count(); + + using SysPeriod = std::chrono::system_clock::period; + Int scaleNumerator = SysPeriod::num * commonDomain.resolution.getDenominator(); + Int scaleDenominator = SysPeriod::den * commonDomain.resolution.getNumerator(); + RangeValue offsetFromCommon = static_cast(epochOffset * scaleNumerator / scaleDenominator); + + // tick_common = tick * multiplier + Int multiplierNumerator = domain.resolution.getNumerator() * commonDomain.resolution.getDenominator(); + Int multiplierDenominator = domain.resolution.getDenominator() * commonDomain.resolution.getNumerator(); + RangeValue startScaledToCommon = + static_cast((value.start / multiplierDenominator) * multiplierNumerator + + (value.start % multiplierDenominator) * multiplierNumerator / multiplierDenominator); + RangeValue endScaledToCommon = + static_cast((value.end / multiplierDenominator) * multiplierNumerator + + (value.end % multiplierDenominator) * multiplierNumerator / multiplierDenominator); + + RangeValue startInCommon = offsetFromCommon + startScaledToCommon; + RangeValue endInCommon = value.end == -1 ? static_cast(-1) : offsetFromCommon + endScaledToCommon; + + return std::make_unique>(commonDomain, RangeType64{startInCommon, endInCommon}); + } + + std::unique_ptr fromCommonDomain(const DomainInfo& regularDomain) override + { + const auto& commonDomain = this->domain; + const auto& valueInCommon = this->value; + + // Offset of regularDomain domain in common domain ticks + Int epochOffset = regularDomain.epoch.time_since_epoch().count() - commonDomain.epoch.time_since_epoch().count(); + + using SysPeriod = std::chrono::system_clock::period; + Int scaleNumerator = SysPeriod::num * commonDomain.resolution.getDenominator(); + Int scaleDenominator = SysPeriod::den * commonDomain.resolution.getNumerator(); + Int offsetFromCommon = epochOffset * scaleNumerator / scaleDenominator; + + RangeValue startScaledToCommon = valueInCommon.start - offsetFromCommon; + RangeValue endScaledToCommon = valueInCommon.end - offsetFromCommon; + + // tick_common = tick * multiplier + Int multiplierNumerator = regularDomain.resolution.getNumerator() * commonDomain.resolution.getDenominator(); + Int multiplierDenominator = regularDomain.resolution.getDenominator() * commonDomain.resolution.getNumerator(); + RangeValue startValue = + static_cast((startScaledToCommon / multiplierNumerator) * multiplierDenominator + + (startScaledToCommon % multiplierNumerator) * multiplierDenominator / multiplierNumerator); + RangeValue endValue = + valueInCommon.end == -1 + ? static_cast(-1) + : static_cast((endScaledToCommon / multiplierNumerator) * multiplierDenominator + + (endScaledToCommon % multiplierNumerator) * multiplierDenominator / multiplierNumerator); + + return std::make_unique>(regularDomain, RangeType64{startValue, endValue}); + } + + void roundUpOnDomainInterval(const RatioPtr& interval) override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + RangeType64 getValue() const + { + return value; + } + +#if !defined(NDEBUG) + virtual std::string asTime() const override + { + using namespace reader; + + std::stringstream ss; + ss << toSysTime(value.start, domain.epoch, domain.resolution); + + return ss.str(); + } +#endif + + int compare(const DomainValue& other) const override + { + const auto* otherImpl = dynamic_cast*>(&other); + if (otherImpl == nullptr) + { + DAQ_THROW_EXCEPTION(InvalidParameterException, "Both DomainValue objects must be of the same type!"); + } + if (otherImpl->domain != this->domain) + DAQ_THROW_EXCEPTION(InvalidParameterException, "Have to compare DomainValue objects in the same domain!"); + + if (this->value.start > otherImpl->value.start) + return 1; + else if (this->value.start == otherImpl->value.start) + return 0; + else // this->value.start < otherImpl->value.start + return -1; + } + +private: + RangeType64 value; +}; + +template <> +class DomainValueImpl final : public DomainValue +{ +public: + explicit DomainValueImpl(const DomainInfo& info, ComplexFloat32 value) + : DomainValue(info) + { + } + + std::unique_ptr toCommonDomain(const DomainInfo& commonDomain) override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + std::unique_ptr fromCommonDomain(const DomainInfo& regularDomain) override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + void roundUpOnDomainInterval(const RatioPtr& interval) override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + ComplexFloat32 getValue() const + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + +#if !defined(NDEBUG) + virtual std::string asTime() const override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } +#endif + + int compare(const DomainValue& other) const override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } +}; + +template <> +class DomainValueImpl final : public DomainValue +{ +public: + explicit DomainValueImpl(const DomainInfo& info, ComplexFloat64 value) + : DomainValue(info) + { + } + + std::unique_ptr toCommonDomain(const DomainInfo& commonDomain) override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + std::unique_ptr fromCommonDomain(const DomainInfo& regularDomain) override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + void roundUpOnDomainInterval(const RatioPtr& interval) override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + ComplexFloat64 getValue() const + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + +#if !defined(NDEBUG) + virtual std::string asTime() const override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } +#endif + + int compare(const DomainValue& other) const override + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } +}; + +END_NAMESPACE_OPENDAQ diff --git a/core/opendaq/reader/include/opendaq/enum_flags.h b/core/opendaq/reader/include/opendaq/enum_flags.h new file mode 100644 index 0000000000..219507fd5d --- /dev/null +++ b/core/opendaq/reader/include/opendaq/enum_flags.h @@ -0,0 +1,73 @@ +/* + * Copyright 2022-2026 openDAQ d.o.o. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +template +class EnumFlags +{ + static_assert(std::is_enum_v, "EnumFlags requires an enum type"); + +public: + using Underlying = std::underlying_type_t; + + constexpr EnumFlags() = default; + + constexpr EnumFlags(Enum value) + : value(static_cast(value)) + { + } + + constexpr bool empty() const + { + return value == 0; + } + + constexpr bool contains(Enum flag) const + { + return (value & static_cast(flag)) != 0; + } + + constexpr void add(Enum flag) + { + value |= static_cast(flag); + } + + constexpr void remove(Enum flag) + { + value &= ~static_cast(flag); + } + + constexpr void set(Enum flag, bool active) + { + if (active) + add(flag); + else + remove(flag); + } + + constexpr void clear() + { + value = 0; + } + + constexpr Underlying raw() const + { + return value; + } + +private: + Underlying value = 0; +}; diff --git a/core/opendaq/reader/include/opendaq/multi_reader_impl.h b/core/opendaq/reader/include/opendaq/multi_reader_impl.h index 13b300b118..df9e09a638 100644 --- a/core/opendaq/reader/include/opendaq/multi_reader_impl.h +++ b/core/opendaq/reader/include/opendaq/multi_reader_impl.h @@ -15,6 +15,7 @@ */ #pragma once #include + #include #include #include @@ -184,10 +185,12 @@ class MultiReaderImpl : public ImplementationOfWeak commonStart; + // std::unique_ptr commonStart; + std::unique_ptr commonDomainStart; std::int64_t requiredCommonSampleRate = -1; std::int64_t commonSampleRate = -1; std::int32_t sampleRateDividerLcm = 1; diff --git a/core/opendaq/reader/include/opendaq/queue_reader.h b/core/opendaq/reader/include/opendaq/queue_reader.h new file mode 100644 index 0000000000..690aa34eb0 --- /dev/null +++ b/core/opendaq/reader/include/opendaq/queue_reader.h @@ -0,0 +1,199 @@ +/* + * Copyright 2022-2026 openDAQ d.o.o. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +BEGIN_NAMESPACE_OPENDAQ + +enum class SignalEventType +{ + NoChange = 0, + ValueChanged, + DomainChanged, + DomainAndValueChanged, + Gap +}; + +class SignalEvent +{ +public: + SignalEvent(const EventPacketPtr& packet); + + static SignalEvent syncGapEvent(Int gapDiff); + + bool merge(const SignalEvent& otherEvent); + SignalEventType getType() const; + const DataDescriptorPtr& getDomainDescriptor() const; + const DataDescriptorPtr& getValueDescriptor() const; + + EventPacketPtr toEventPacket() const; + +private: + explicit SignalEvent(Int gapDiff); + + void updateType(); +private: + SignalEventType eventType; + DataDescriptorPtr domainDescriptor; + DataDescriptorPtr valueDescriptor; + Int gapDiff; +}; + + +enum class AdvanceResult +{ + Success = 0, + NeedMoreData, + DomainChanged, + OvershotError, + Error +}; + +enum class QueueReaderIssue : uint32_t +{ + None = 0, + ValueTypesNotConvertible = 1 << 0, + DomainTypesNotConvertible = 1 << 1, + UnsupportedDomainRule = 1 << 2, + OriginParsingFailed = 1 << 3, + DomainUnitInvalid = 1 << 4 +}; + +class QueueReader +{ +public: + explicit QueueReader(const InputPortConfigPtr& port, // Consider using Connection instead + SampleType valueReadType, + SampleType domainReadType, + ReadMode mode, + const LoggerComponentPtr& logger, + bool globalIdFromSignal); + +public: + DomainInfo getDomainInfo(); + std::unique_ptr getFirstSampleDomainValue(); + AdvanceResult advanceToDomainValue(const DomainValue* domainValue); + Int getSampleRate(); + + void dropOutdatedPacketSegments(); + + /** + * @brief Get the Available Samples in common rate equivalent + * + * @return SizeT Available samples multiplied by the sample rate divider. + */ + SizeT getAvailableSamples(); + + bool hasPendingEvents(); + EventPacketPtr popFrontEvent(); + + bool isValid(); + + void domainChangeHandled(); + void updateConnection(); + + void setSampleRateDivider(SizeT divider); + SizeT getSampleRateDivider() const; + + /** + * @brief Read common rate equivalent samples into the buffer. There will be nativeSamples = count / sampleRateDivider + * samples read from the packets into the buffer. + * + * @param buffer Buffer that has capacity of at least count / sampleRateDivider + * @param count Desired sample count in common rate equivalent. + * @return AdvanceResult + */ + AdvanceResult read(void* valueBuffer, void* domainBuffer, SizeT* count); + AdvanceResult skip(SizeT* count); + + /** + * @brief Drop the current data segment, if there are fewer than samplesInBlock samples available (common rate equivalent). + * + * For example: If there are 3 samples available before next event, samplesInBlock=10 (this is dividerLCM in terms of multireading) + * and divider for the queue reader is 2, then 5 native samples are required as minimum aligned read. Since 3 < 5, we drop the 3 samples + * and create a synchronization GAP event in the event queue. + * + * @param samplesInBlock Number of samples (common rate equivalent). + * @return true If samples were dropped. + * @return false If samples were not dropped - data segment is long enough or there is no event in the queue to end the segment. + */ + bool dropLeftoverSegment(SizeT samplesInBlock); + +private: + void drainConnection(); + void adoptPackets(); + void consumeLeadingEventPackets(); + + SizeT getAvailableSamplesNative(); + AdvanceResult readNative(void* valueBuffer, void* domainBuffer, SizeT* count); + + void checkConnection() const; + + SignalEventType addEncounteredEvent(const EventPacketPtr& packet); + void addToEventQueue(SignalEvent&& event); + void parseDomainDescriptor(); + void parseValueDescriptor(); + void parseCachedDescriptors(); + size_t getNumberOfEventPacketsInQueue(); + bool dropUntilEvent(); + +private: + std::deque packets; + std::deque events; + + SizeT readingPosition = 0; + + InputPortConfigPtr port; + ConnectionPtr connection; + + LoggerComponentPtr loggerComponent; + + EnumFlags issues; + + ReadMode readMode; + + struct TypedReadingContext + { + SampleType domainIn; + SampleType domainOut; + ReadLayout domainLayout; + DomainInfo domainInfo; + FunctionPtr domainTransform = nullptr; + + SampleType valueIn; + SampleType valueOut; + ReadLayout valueLayout; + FunctionPtr valueTransform = nullptr; + }; + TypedReadingContext typeCtx; + + Int sampleRate = -1; + Int packetDelta{0}; + bool domainChanged = false; + + SizeT sampleRateDivider = 1; +}; + +END_NAMESPACE_OPENDAQ \ No newline at end of file diff --git a/core/opendaq/reader/include/opendaq/reader_utils.h b/core/opendaq/reader/include/opendaq/reader_utils.h index 7b3f1338f9..d817be9746 100644 --- a/core/opendaq/reader/include/opendaq/reader_utils.h +++ b/core/opendaq/reader/include/opendaq/reader_utils.h @@ -21,6 +21,7 @@ #include #include #include +#include BEGIN_NAMESPACE_OPENDAQ @@ -67,17 +68,36 @@ namespace reader if (epoch.find('T') == std::string::npos) { // If no time, assume Midnight UTC - epoch += "T00:00:00+00:00"; + epoch += "T00:00:00+0000"; } - else if (epoch[epoch.size() - 1] == 'Z') + else if (!epoch.empty() && epoch.back() == 'Z') { // If time-zone marked as "Zulu" (UTC) replace with offset - epoch = epoch.erase(epoch.size() - 1) + "+00:00"; + epoch.pop_back(); + epoch += "+0000"; } - else if (epoch.find('+') == std::string::npos) + else { - // If not time-zone offset assume UTC - epoch += "+00:00"; + // Search after T to ignore any '-' character in the date portion + const auto timeStart = epoch.find('T'); + const auto plusPos = epoch.find('+', timeStart); + const auto minusPos = epoch.find('-', timeStart); + + const auto offsetPos = plusPos != std::string::npos ? plusPos : minusPos; + + if (offsetPos == std::string::npos) + { + // If not time-zone offset assume UTC + epoch += "+0000"; + } + else + { + const auto colonPos = epoch.find(':', offsetPos); + if (colonPos != std::string::npos) + { + epoch.erase(colonPos, 1); + } + } } return epoch; @@ -94,6 +114,27 @@ namespace reader return epoch; } + + inline std::optional tryParseEpoch(const std::string& origin) + { + std::chrono::system_clock::time_point epoch; + + std::istringstream epochString(fixupIso8601(origin)); + date::from_stream(epochString, "%FT%T%z", epoch); + + if (epochString.fail()) + { + return std::nullopt; + } + + epochString >> std::ws; // strip trailing whitespace + if (epochString.peek() != std::istringstream::traits_type::eof()) + { + return std::nullopt; + } + + return epoch; + } namespace detail { diff --git a/core/opendaq/reader/include/opendaq/signal_reader.h b/core/opendaq/reader/include/opendaq/signal_reader.h index f1a9c42f14..24438ccb75 100644 --- a/core/opendaq/reader/include/opendaq/signal_reader.h +++ b/core/opendaq/reader/include/opendaq/signal_reader.h @@ -14,12 +14,13 @@ * limitations under the License. */ #pragma once +#include #include #include -#include #include #include #include +#include #include @@ -60,7 +61,7 @@ struct SignalReader */ SizeT getAvailable(bool acrossDescriptorChanges) const; void handleDescriptorChanged(const EventPacketPtr& eventPacket); - bool trySetDomainSampleType(const daq::DataPacketPtr& domainPacket) const; + bool trySetDomainSampleType(const daq::DataPacketPtr& domainPacket); void setCommonSampleRate(const std::int64_t commonSampleRate); void prepare(void* outValues, SizeT count); @@ -71,7 +72,16 @@ struct SignalReader /** * @brief Returns the tick of the first available sample in maxResolution units and relative to the minimum epoch. */ - std::unique_ptr readStartDomain(); + std::unique_ptr readDomainStart(); + const DomainInfo& getDomainInfo() const; + SampleType getValueReadType() const; + + void setValueTransformFunction(FunctionPtr transform); + void setDomainTransformFunction(FunctionPtr transform); + + FunctionPtr getValueTransformFunction() const; + FunctionPtr getDomainTransformFunction() const; + /** * @brief Dequeues first datapacket if available and returns true if first packet is Event. * @@ -83,7 +93,7 @@ struct SignalReader bool isFirstPacketEvent(); EventPacketPtr readUntilNextDataPacket(); bool skipUntilLastEventPacket(); - bool sync(const Comparable& commonStart, std::chrono::system_clock::rep* firstSampleAbsoluteTimestamp = nullptr); + bool sync(const DomainValue* commonStart, std::chrono::system_clock::rep* firstSampleAbsoluteTimestamp = nullptr); ErrCode readPackets(); ErrCode readPacketData(); @@ -97,9 +107,6 @@ struct SignalReader LoggerComponentPtr loggerComponent; - std::unique_ptr valueReader; - std::unique_ptr domainReader; - InputPortConfigPtr port; ConnectionPtr connection; @@ -118,6 +125,25 @@ struct SignalReader NumberPtr packetDelta {0}; std::chrono::system_clock::rep cachedFirstTimestamp; + +private: + + bool onValueDescriptorUpdate(const DataDescriptorPtr& valueDescriptor); + bool onDomainDescriptorUpdate(const DataDescriptorPtr& domainDescriptor); + struct TypedReadingContext + { + SampleType domainIn; + SampleType domainOut; + ReadLayout domainLayout; + DomainInfo domainInfo; + FunctionPtr domainTransform = nullptr; + + SampleType valueIn; + SampleType valueOut; + ReadLayout valueLayout; + FunctionPtr valueTransform = nullptr; + }; + TypedReadingContext trContext; }; END_NAMESPACE_OPENDAQ diff --git a/core/opendaq/reader/include/opendaq/typed_reading_utils.h b/core/opendaq/reader/include/opendaq/typed_reading_utils.h new file mode 100644 index 0000000000..c92e277a63 --- /dev/null +++ b/core/opendaq/reader/include/opendaq/typed_reading_utils.h @@ -0,0 +1,64 @@ +/* + * Copyright 2022-2025 openDAQ d.o.o. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include +#include +#include + + +BEGIN_NAMESPACE_OPENDAQ + +struct ReadLayout +{ + DataDescriptorPtr descriptor = nullptr; + SizeT rawSampleSize = 0; + SizeT valuesPerSample = 1; +}; + +class TypedReadingUtils +{ +public: + static ReadLayout createReadLayout(const DataDescriptorPtr& descriptor); + + static bool isSampleTypeConvertible(SampleType in, SampleType out, bool isDomain); + + static std::unique_ptr readDomainValue(SampleType in, + SampleType out, + const ReadLayout& readLayout, + const DataPacketPtr& domainPacket, + SizeT index, + const DomainInfo& domainInfo); + +static SizeT findDomainValue(SampleType in, + SampleType out, + const ReadLayout& readLayout, + const DataPacketPtr& domainPacket, + const DomainValue* target, + std::chrono::system_clock::rep* firstSampleAbsoluteTime = nullptr); + + static ErrCode readData(SampleType in, + SampleType out, + bool isDomain, + const ReadLayout& readLayout, + void* inputBuffer, + SizeT offset, + void** outputBuffer, + SizeT count, + const FunctionPtr transform = nullptr); +}; + +END_NAMESPACE_OPENDAQ \ No newline at end of file diff --git a/core/opendaq/reader/src/CMakeLists.txt b/core/opendaq/reader/src/CMakeLists.txt index 5fa45f3d81..6571528fde 100644 --- a/core/opendaq/reader/src/CMakeLists.txt +++ b/core/opendaq/reader/src/CMakeLists.txt @@ -83,11 +83,14 @@ function(create_component_source_groups_${BASE_NAME}) ${SDK_HEADERS_DIR}/time_reader.h ${SDK_HEADERS_DIR}/read_info.h ${SDK_HEADERS_DIR}/typed_reader.h + ${SDK_HEADERS_DIR}/typed_reading_utils.h + ${SDK_HEADERS_DIR}/domain_value.h ${SDK_HEADERS_DIR}/reader_impl.h ${SDK_HEADERS_DIR}/reader_status_impl.h ${SDK_SRC_DIR}/reader_impl.cpp ${SDK_SRC_DIR}/reader_status_impl.cpp ${SDK_SRC_DIR}/typed_reader.cpp + ${SDK_SRC_DIR}/typed_reading_utils.cpp ) source_group("reader//stream" FILES @@ -131,6 +134,7 @@ function(create_component_source_groups_${BASE_NAME}) ${SDK_HEADERS_DIR}/multi_reader_builder_impl.h ${SDK_HEADERS_DIR}/multi_reader.h ${SDK_HEADERS_DIR}/signal_reader.h + ${SDK_HEADERS_DIR}/queue_reader.h ${SDK_HEADERS_DIR}/multi_reader_impl.h ${SDK_HEADERS_DIR}/multi_typed_reader.h ${SDK_HEADERS_DIR}/reader_domain_info.h @@ -138,6 +142,7 @@ function(create_component_source_groups_${BASE_NAME}) ${SDK_SRC_DIR}/multi_reader_impl.cpp ${SDK_SRC_DIR}/multi_reader_builder_impl.cpp ${SDK_SRC_DIR}/signal_reader.cpp + ${SDK_SRC_DIR}/queue_reader.cpp ) endfunction() @@ -146,6 +151,8 @@ set(SRC_PublicHeaders_Component reader_errors.h reader_exceptions.h typed_reader.h + typed_reading_utils.h + domain_value.h time_reader.h read_info.h reader_utils.h @@ -165,8 +172,10 @@ set(SRC_PrivateHeaders_Component multi_reader_builder_impl.h multi_typed_reader.h signal_reader.h + queue_reader.h reader_status_impl.h reader_impl.h + enum_flags.h PARENT_SCOPE ) @@ -181,9 +190,11 @@ set(SRC_Cpp_Component reader_status_impl.cpp reader_impl.cpp typed_reader.cpp + typed_reading_utils.cpp multi_reader_impl.cpp multi_reader_builder_impl.cpp signal_reader.cpp + queue_reader.cpp reader.natvis PARENT_SCOPE ) diff --git a/core/opendaq/reader/src/multi_reader_impl.cpp b/core/opendaq/reader/src/multi_reader_impl.cpp index e991df5c0e..08c85c2d79 100644 --- a/core/opendaq/reader/src/multi_reader_impl.cpp +++ b/core/opendaq/reader/src/multi_reader_impl.cpp @@ -24,7 +24,7 @@ using namespace std::chrono; using Milliseconds = duration; template <> -struct fmt::formatter : ostream_formatter +struct fmt::formatter : ostream_formatter { }; @@ -575,17 +575,19 @@ void MultiReaderImpl::setStartInfo() if (signal.unused) continue; - if (signal.domainInfo.epoch < minEpoch) + auto& domainInfo = signal.getDomainInfo(); + if (domainInfo.epoch < minEpoch) { - minEpoch = signal.domainInfo.epoch; + minEpoch = domainInfo.epoch; } - if (static_cast(signal.domainInfo.resolution) < static_cast(maxResolution)) + if (static_cast(domainInfo.resolution) < static_cast(maxResolution)) { - maxResolution = signal.domainInfo.resolution; + maxResolution = domainInfo.resolution; } } + commonDomain = DomainInfo{minEpoch, maxResolution}; readResolution = maxResolution; readOrigin = date::format("%FT%TZ", minEpoch); @@ -624,7 +626,7 @@ ErrCode MultiReaderImpl::getValueReadType(SampleType* sampleType) // value read type may differ from what was configured (e. g. Undefined -> Int64). // The SignalReader will instantiate the appropriate type reader when descriptors change. // Shouldn't be relied on if different signals are read simultaneously. - *sampleType = signals.front().valueReader->getReadType(); + *sampleType = signals.front().getValueReadType(); else *sampleType = valueReadType; return OPENDAQ_SUCCESS; @@ -644,7 +646,7 @@ ErrCode MultiReaderImpl::setValueTransformFunction(IFunction* transform) for (auto& signal : signals) { - signal.valueReader->setTransformFunction(transform); + signal.setValueTransformFunction(transform); } return OPENDAQ_SUCCESS; @@ -656,7 +658,7 @@ ErrCode MultiReaderImpl::setDomainTransformFunction(IFunction* transform) for (auto& signal : signals) { - signal.domainReader->setTransformFunction(transform); + signal.setDomainTransformFunction(transform); } return OPENDAQ_SUCCESS; @@ -739,7 +741,7 @@ ErrCode INTERFACE_FUNC MultiReaderImpl::removeInput(IString* globalId) signals.erase(it); // Reset common start to avoid holding a reference to the deleted signal reader. - commonStart = nullptr; + commonDomainStart = nullptr; portsConnected = allPortsConnected(); if (!portsConnected) @@ -1472,27 +1474,21 @@ void MultiReaderImpl::readDomainStart() // => SignalReader should be able to handle two domain settings - one native to the signal it is reading and // another, "common", that will be set from Multireader parent. Ideally, getting these common domain information, // it should be trivial to compare starts. - auto sigStart = signal.readStartDomain(); - if (!commonStart || *commonStart < *sigStart) + // auto sigStart = signal.readStartDomain(); + auto sigStart = signal.readDomainStart(); + auto sigStartInCommonDomain = sigStart->toCommonDomain(commonDomain); + if (!commonDomainStart || *commonDomainStart < *sigStartInCommonDomain) { - commonStart = std::move(sigStart); + commonDomainStart = std::move(sigStartInCommonDomain); } } LOG_T("---"); - LOG_T("DomainStart: {}", *commonStart); + LOG_T("DomainStart: {}", *commonDomainStart); - if (startOnFullUnitOfDomain) - { - commonStart->roundUpOnUnitOfDomain(); - LOG_T("Rounded DomainStart: {}", *commonStart); - } - else - { - const RatioPtr interval = Ratio(sampleRateDividerLcm, commonSampleRate).simplify(); - commonStart->roundUpOnDomainInterval(interval); - LOG_T("Aligned DomainStart: {}", *commonStart); - } + const RatioPtr interval = startOnFullUnitOfDomain ? Ratio(1, 1) : Ratio(sampleRateDividerLcm, commonSampleRate).simplify(); + commonDomainStart->roundUpOnDomainInterval(interval); + LOG_T("Aligned DomainStart: {}", *commonDomainStart); } void MultiReaderImpl::sync() @@ -1507,7 +1503,8 @@ void MultiReaderImpl::sync() continue; system_clock::rep firstSampleAbsoluteTime; - synced = signal.sync(*commonStart, &firstSampleAbsoluteTime) && synced; + auto startInNativeDomain = commonDomainStart->fromCommonDomain(signal.getDomainInfo()); + synced = signal.sync(startInNativeDomain.get(), &firstSampleAbsoluteTime) && synced; if (synced) { @@ -1564,9 +1561,10 @@ ErrCode MultiReaderImpl::getOffset(void* domainStart) { OPENDAQ_PARAM_NOT_NULL(domainStart); - if (commonStart) + if (commonDomainStart) { - commonStart->getValue(domainStart); + // TODO: What can this be possibly used for? + // commonStart->getValue(domainStart); return OPENDAQ_SUCCESS; } @@ -1586,7 +1584,8 @@ ErrCode MultiReaderImpl::getIsSynchronized(Bool* isSynchronized) { OPENDAQ_PARAM_NOT_NULL(isSynchronized); - *isSynchronized = static_cast(commonStart); + // TODO: There is a more comlpex answer to this + *isSynchronized = commonDomainStart != nullptr; return OPENDAQ_SUCCESS; } @@ -1634,7 +1633,7 @@ ErrCode MultiReaderImpl::getValueTransformFunction(IFunction** transform) return OPENDAQ_ERR_INVALIDSTATE; } - *transform = signals.front().valueReader->getTransformFunction().addRefAndReturn(); + *transform = signals.front().getValueTransformFunction().addRefAndReturn(); return OPENDAQ_SUCCESS; } @@ -1650,7 +1649,7 @@ ErrCode MultiReaderImpl::getDomainTransformFunction(IFunction** transform) return OPENDAQ_ERR_INVALIDSTATE; } - *transform = signals.front().domainReader->getTransformFunction().addRefAndReturn(); + *transform = signals.front().getDomainTransformFunction().addRefAndReturn(); return OPENDAQ_SUCCESS; } diff --git a/core/opendaq/reader/src/queue_reader.cpp b/core/opendaq/reader/src/queue_reader.cpp new file mode 100644 index 0000000000..652928ad88 --- /dev/null +++ b/core/opendaq/reader/src/queue_reader.cpp @@ -0,0 +1,767 @@ +#include + +#include +#include + +BEGIN_NAMESPACE_OPENDAQ + +SignalEvent::SignalEvent(const EventPacketPtr& packet) + : eventType(SignalEventType::NoChange) + , domainDescriptor(nullptr) + , valueDescriptor(nullptr) + , gapDiff(0) +{ + if (packet.getEventId() == event_packet_id::IMPLICIT_DOMAIN_GAP_DETECTED) + { + eventType = SignalEventType::Gap; + gapDiff = packet.getParameters().get(event_packet_param::GAP_DIFF); + } + else + { + const auto [valueDescChanged, domainDescChanged, newValueDescriptor, newDomainDescriptor] = parseDataDescriptorEventPacket(packet); + domainDescriptor = newDomainDescriptor; + valueDescriptor = newValueDescriptor; + updateType(); + } +} + +SignalEvent::SignalEvent(Int gapDiff) + : eventType(SignalEventType::Gap) + , domainDescriptor(nullptr) + , valueDescriptor(nullptr) + , gapDiff(gapDiff) +{ +} + +SignalEvent SignalEvent::syncGapEvent(Int gapDiff) +{ + return SignalEvent(gapDiff); +} + +void SignalEvent::updateType() +{ + if (eventType == SignalEventType::Gap) + return; + + if (domainDescriptor.assigned() and valueDescriptor.assigned()) + { + eventType = SignalEventType::DomainAndValueChanged; + } + else if (domainDescriptor.assigned()) + { + eventType = SignalEventType::DomainChanged; + } + else if (valueDescriptor.assigned()) + { + eventType = SignalEventType::ValueChanged; + } + else + { + eventType = SignalEventType::NoChange; + } +} + +bool SignalEvent::merge(const SignalEvent& other) +{ + if (this->eventType != SignalEventType::Gap && other.eventType == SignalEventType::Gap) + return false; + if (this->eventType == SignalEventType::Gap && other.eventType != SignalEventType::Gap) + return false; + + if (other.domainDescriptor.assigned()) + domainDescriptor = other.domainDescriptor; + if (other.valueDescriptor.assigned()) + valueDescriptor = other.valueDescriptor; + updateType(); + return true; +} + +SignalEventType SignalEvent::getType() const +{ + return eventType; +} + +const DataDescriptorPtr& SignalEvent::getDomainDescriptor() const +{ + return domainDescriptor; +} + +const DataDescriptorPtr& SignalEvent::getValueDescriptor() const +{ + return valueDescriptor; +} + +EventPacketPtr SignalEvent::toEventPacket() const +{ + if (eventType == SignalEventType::Gap) + { + return ImplicitDomainGapDetectedEventPacket(gapDiff); + } + else + { + return DataDescriptorChangedEventPacket(descriptorToEventPacketParam(valueDescriptor), descriptorToEventPacketParam(domainDescriptor)); + } +} + +QueueReader::QueueReader(const InputPortConfigPtr& port, // Consider using Connection instead + SampleType valueReadType, + SampleType domainReadType, + ReadMode mode, + const LoggerComponentPtr& logger, + bool globalIdFromSignal) // TODO + : port(port) + , connection(port.getConnection()) + , readMode(mode) + , loggerComponent(logger) +{ + typeCtx.domainIn = SampleType::Undefined; + typeCtx.domainOut = domainReadType; + typeCtx.valueIn = SampleType::Undefined; + typeCtx.valueOut = mode == ReadMode::RawValue ? SampleType::Undefined : valueReadType; +} + +void QueueReader::adoptPackets() +{ + // Take ownership of all packets + PacketPtr packet = connection.dequeue(); + while (packet.assigned()) + { + packets.push_back(std::move(packet)); + packet = connection.dequeue(); + } +} + +DomainInfo QueueReader::getDomainInfo() +{ + checkConnection(); + + drainConnection(); + return typeCtx.domainInfo; +} + +std::unique_ptr QueueReader::getFirstSampleDomainValue() +{ + checkConnection(); + drainConnection(); + + if (packets.empty() || packets.front().getType() != PacketType::Data) + { + return nullptr; + } + + DataPacketPtr domainPacket = packets.front().asPtr(true).getDomainPacket(); + if (!domainPacket.assigned()) + { + DAQ_THROW_EXCEPTION(InvalidStateException, "Packet must have a domain packet assigned!"); + } + + return TypedReadingUtils::readDomainValue( + typeCtx.domainIn, typeCtx.domainOut, typeCtx.domainLayout, domainPacket, readingPosition, typeCtx.domainInfo); +} + +AdvanceResult QueueReader::advanceToDomainValue(const DomainValue* domainValue) +{ + // TODO: Add first timestamp mechanism for sync tolerance checking + checkConnection(); + drainConnection(); + + SignalEventType signalChange = SignalEventType::NoChange; + + bool found = false; + SizeT end = 0; + for (auto& packet : packets) + { + if (packet.getType() == PacketType::Data) + { + DataPacketPtr domainPacket = packet.asPtr(true).getDomainPacket(); + + SizeT index = TypedReadingUtils::findDomainValue( + typeCtx.domainIn, typeCtx.domainOut, typeCtx.domainLayout, domainPacket, domainValue, nullptr /*TODO*/); + + if (index != static_cast(-1)) + { + if (index < readingPosition) + { + return AdvanceResult::OvershotError; + } + readingPosition = index; + found = true; + break; + } + + readingPosition = 0; + ++end; + continue; + } + else if (packet.getType() == PacketType::Event) + { + auto eventPacket = packet.asPtr(true); + signalChange = addEncounteredEvent(eventPacket); + ++end; + + if (signalChange == SignalEventType::DomainChanged || signalChange == SignalEventType::DomainAndValueChanged || + signalChange == SignalEventType::Gap) + { + break; + } + continue; + } + else + { + // Unexpected packet type encountered. + // Packet should be removed and sync is not successful. + signalChange = SignalEventType::DomainChanged; + ++end; + break; + } + } + + packets.erase(packets.begin(), packets.begin() + end); + + switch (signalChange) + { + case SignalEventType::DomainChanged: + case SignalEventType::DomainAndValueChanged: + case SignalEventType::Gap: + return AdvanceResult::DomainChanged; + default: + break; + } + + return found ? AdvanceResult::Success : AdvanceResult::NeedMoreData; +} + +Int QueueReader::getSampleRate() +{ + checkConnection(); + drainConnection(); + + return sampleRate; +} + +void QueueReader::consumeLeadingEventPackets() +{ + size_t end = 0; + for (const auto& packet : packets) + { + auto packetType = packet.getType(); + if (packetType == PacketType::Data) + { + break; + } + + EventPacketPtr eventPacket = packet.asPtr(true); + addEncounteredEvent(eventPacket); + + ++end; + } + packets.erase(packets.begin(), packets.begin() + end); +} + +void QueueReader::checkConnection() const +{ + if (!connection.assigned()) + DAQ_THROW_EXCEPTION(InvalidOperationException, "Connection must be assigned for this operation."); +} + +void QueueReader::dropOutdatedPacketSegments() +{ + checkConnection(); + drainConnection(); + + while (getNumberOfEventPacketsInQueue() >= 2) + { + auto foundEvent = dropUntilEvent(); + assert(foundEvent && "Event should have been found."); + consumeLeadingEventPackets(); + } + dropUntilEvent(); + consumeLeadingEventPackets(); +} + +SizeT QueueReader::getAvailableSamplesNative() +{ + checkConnection(); + drainConnection(); + + SizeT count = 0; + SizeT packetReadingPosition = readingPosition; + for (const auto& packet : packets) + { + if (packet.getType() != PacketType::Data) + break; + + DataPacketPtr dataPacket = packet.asPtr(true); + count += dataPacket.getSampleCount() - packetReadingPosition; + + // Only first packet may have non-zero reading position + packetReadingPosition = 0; + } + return count; +} + +SizeT QueueReader::getAvailableSamples() +{ + return getAvailableSamplesNative() * sampleRateDivider; +} + +bool QueueReader::hasPendingEvents() +{ + checkConnection(); + drainConnection(); + return !events.empty(); +} + +EventPacketPtr QueueReader::popFrontEvent() +{ + checkConnection(); + drainConnection(); + if (events.empty()) + return nullptr; + + auto eventPacket = events.front().toEventPacket(); + events.pop_front(); + return eventPacket; +} + +bool QueueReader::isValid() +{ + if (!connection.assigned()) + return false; + + drainConnection(); + return issues.empty(); +} + +void QueueReader::domainChangeHandled() +{ + domainChanged = false; +} + +void QueueReader::updateConnection() +{ + connection = port.getConnection(); + drainConnection(); +} + +void QueueReader::setSampleRateDivider(SizeT divider) +{ + if (divider == 0) + { + DAQ_THROW_EXCEPTION(InvalidParameterException, "Sample rate divider must not be 0."); + } + sampleRateDivider = divider; +} + +SizeT QueueReader::getSampleRateDivider() const +{ + return sampleRateDivider; +} + +AdvanceResult QueueReader::read(void* valueBuffer, void* domainBuffer, SizeT* count) +{ + if (count == nullptr) + return AdvanceResult::Error; + + if (*count % sampleRateDivider != 0) + { + // Enforce reading in units of common samples + *count = 0; + return AdvanceResult::Error; + } + SizeT nativeCount = *count / sampleRateDivider; + AdvanceResult result = readNative(valueBuffer, domainBuffer, &nativeCount); + *count = nativeCount * sampleRateDivider; + return result; +} + +AdvanceResult QueueReader::readNative(void* valueBuffer, void* domainBuffer, SizeT* count) +{ + if (count == nullptr) + return AdvanceResult::Error; + + if (*count == 0) + return AdvanceResult::Success; + + if (hasPendingEvents()) + { + *count = 0; + return AdvanceResult::Error; + } + + const SizeT requested = *count; + SizeT remainingToRead = requested; + void* valuePtr = valueBuffer; + void* domainPtr = domainBuffer; + + bool returnError = false; + SizeT end = 0; + for (auto& packet : packets) + { + if (packet.getType() != PacketType::Data) + { + // The user of this class should read <= available samples, so event is not encountered + returnError = true; + break; + } + + DataPacketPtr dataPacket = packet.asPtr(true); + SizeT sampleCount = dataPacket.getSampleCount(); + + if (readingPosition > sampleCount) + { + returnError = true; + break; + } + + SizeT remainingInPacket = sampleCount - readingPosition; + SizeT toRead = std::min(remainingToRead, remainingInPacket); + + if (toRead == 0) + { + readingPosition = 0; + ++end; + continue; + } + + if (valuePtr != nullptr) + { + void* valueData = nullptr; + switch (readMode) + { + case ReadMode::RawValue: + case ReadMode::Unscaled: + valueData = dataPacket.getRawData(); + break; + case ReadMode::Scaled: + default: + valueData = dataPacket.getData(); + break; + } + + ErrCode errCode = TypedReadingUtils::readData(typeCtx.valueIn, + typeCtx.valueOut, + false, + typeCtx.valueLayout, + valueData, + readingPosition, + &valuePtr, + toRead, + typeCtx.valueTransform); + if (!OPENDAQ_SUCCEEDED(errCode)) + throwExceptionFromErrorCode(errCode, getErrorInfoMessage(errCode, true)); + } + + if (domainPtr != nullptr) + { + auto domainPacket = dataPacket.getDomainPacket(); + if (!domainPacket.assigned()) + DAQ_THROW_EXCEPTION(NotSupportedException, "Domain packet must be assigned."); + + ErrCode errCode = TypedReadingUtils::readData(typeCtx.domainIn, + typeCtx.domainOut, + true, + typeCtx.domainLayout, + domainPacket.getData(), + readingPosition, + &domainPtr, + toRead, + typeCtx.domainTransform); + + if (!OPENDAQ_SUCCEEDED(errCode)) + throwExceptionFromErrorCode(errCode, getErrorInfoMessage(errCode, true)); + } + + remainingToRead -= toRead; + + if (remainingToRead == 0) + { + readingPosition += toRead; + if (readingPosition == sampleCount) + { + readingPosition = 0; + ++end; + } + break; + } + + readingPosition = 0; + ++end; + } + packets.erase(packets.begin(), packets.begin() + end); + *count = requested - remainingToRead; + + if (returnError) + return AdvanceResult::Error; + + // Parse trailing events + if (readingPosition == 0) + consumeLeadingEventPackets(); + + return remainingToRead == 0 ? AdvanceResult::Success : AdvanceResult::NeedMoreData; +} + +AdvanceResult QueueReader::skip(SizeT* count) +{ + return read(nullptr, nullptr, count); +} + +void QueueReader::drainConnection() +{ + if (!connection.assigned()) + return; + + if (!connection.peek().assigned()) + return; + + adoptPackets(); + consumeLeadingEventPackets(); +} + +bool QueueReader::dropLeftoverSegment(SizeT samplesInBlock) +{ + if (samplesInBlock % sampleRateDivider != 0) + { + DAQ_THROW_EXCEPTION(InvalidStateException, "Aligned block size must be divisible by all signal dividers."); + } + + if (hasPendingEvents()) + { + DAQ_THROW_EXCEPTION(InvalidStateException, "Events must be handled before dropping leftover segments."); + } + + // No events in the queue, this segment has not been ended - mustn't drop + if (getNumberOfEventPacketsInQueue() == 0) + return false; + + + const SizeT requiredNativeSamples = samplesInBlock / sampleRateDivider; + size_t availableNativeSamples = getAvailableSamplesNative(); + + if (availableNativeSamples >= requiredNativeSamples) + return false; + + dropUntilEvent(); + addToEventQueue(SignalEvent::syncGapEvent(availableNativeSamples)); + consumeLeadingEventPackets(); // Transition to new segment + return true; +} + +SignalEventType QueueReader::addEncounteredEvent(const EventPacketPtr& packet) +{ + auto event = SignalEvent(packet); + auto eventType = event.getType(); + + switch (eventType) + { + case SignalEventType::DomainChanged: + typeCtx.domainLayout.descriptor = event.getDomainDescriptor(); + break; + case SignalEventType::ValueChanged: + typeCtx.valueLayout.descriptor = event.getValueDescriptor(); + break; + case SignalEventType::DomainAndValueChanged: + typeCtx.domainLayout.descriptor = event.getDomainDescriptor(); + typeCtx.valueLayout.descriptor = event.getValueDescriptor(); + break; + default: + break; + } + parseCachedDescriptors(); + addToEventQueue(std::move(event)); + return eventType; +} + +void QueueReader::addToEventQueue(SignalEvent&& event) +{ + bool eventMerged = false; + if (!events.empty()) + { + // Attempt merging with the last event and add to list if merge not possible + eventMerged = events.back().merge(event); + } + if (!eventMerged) + events.push_back(event); +} + +void QueueReader::parseDomainDescriptor() +{ + auto& descriptor = typeCtx.domainLayout.descriptor; + if (!descriptor.assigned()) + return; + + // Type conversion + const auto postScaling = descriptor.getPostScaling(); + if (!postScaling.assigned() || readMode == ReadMode::Scaled) + { + typeCtx.domainIn = descriptor.getSampleType(); + } + else + { + typeCtx.domainIn = postScaling.getInputSampleType(); + } + + typeCtx.domainLayout.rawSampleSize = descriptor.getRawSampleSize(); + auto dimensions = descriptor.getDimensions(); + if (dimensions.assigned() && dimensions.getCount() == 1) + { + typeCtx.domainLayout.valuesPerSample = dimensions[0].getSize(); + } + + typeCtx.domainInfo = DomainInfo::fromDescriptor(descriptor); + + bool domainTypesConvertible = TypedReadingUtils::isSampleTypeConvertible(typeCtx.domainIn, typeCtx.domainOut, true); + issues.set(QueueReaderIssue::DomainTypesNotConvertible, !domainTypesConvertible); + // END Type Conversion + + // Resolution and origin + auto newResolution = descriptor.getTickResolution(); + if (typeCtx.domainInfo.resolution != newResolution) + { + typeCtx.domainInfo.resolution = newResolution; + domainChanged = true; + } + + std::string origin = descriptor.getOrigin(); + auto newOrigin = reader::tryParseEpoch(origin); + if (newOrigin.has_value() && typeCtx.domainInfo.epoch != newOrigin.value()) + { + typeCtx.domainInfo.epoch = newOrigin.value(); + domainChanged = true; + } + issues.set(QueueReaderIssue::OriginParsingFailed, !newOrigin.has_value()); + // END Resolution and origin + + // Sample rate and delta + { + std::int64_t newSampleRate = 0; + + NumberPtr delta = 1; + auto rule = descriptor.getRule(); + const bool ruleIsLinear = rule.assigned() && rule.getType() == DataRuleType::Linear; + + if (ruleIsLinear) + { + delta = rule.getParameters()["delta"]; + } + + double sr = static_cast(typeCtx.domainInfo.resolution.getDenominator()) / + (static_cast(typeCtx.domainInfo.resolution.getNumerator()) * delta.getFloatValue()); + + const bool deltaIsInteger = (delta.getFloatValue() == static_cast(delta.getIntValue())); + const bool sampleRateIsInteger = (sr == static_cast(static_cast(sr))); + + newSampleRate = static_cast(sr); + + if (sampleRate != newSampleRate) + { + sampleRate = newSampleRate; + domainChanged = true; + } + + if (packetDelta != delta.getIntValue()) + { + packetDelta = delta.getIntValue(); + domainChanged = true; + } + + issues.set(QueueReaderIssue::UnsupportedDomainRule, !ruleIsLinear || !deltaIsInteger || !sampleRateIsInteger); + } + // END Sample rate and delta + + // Unit and quantity + bool domainIsTimeInSeconds = false; + do + { + auto domainUnit = descriptor.getUnit(); + if (!domainUnit.assigned()) + break; + + const auto domainQuantity = domainUnit.getQuantity(); + if (!domainQuantity.assigned() || domainQuantity.getLength() == 0) + break; + if (domainQuantity != "time") + break; + + const auto domainUnitSymbol = domainUnit.getSymbol(); + if (domainUnitSymbol != "s") + break; + + domainIsTimeInSeconds = true; + } while (false); + issues.set(QueueReaderIssue::DomainUnitInvalid, !domainIsTimeInSeconds); + // END Unit and quantity +} + +void QueueReader::parseValueDescriptor() +{ + auto& descriptor = typeCtx.valueLayout.descriptor; + + auto postScaling = descriptor.getPostScaling(); + if (!postScaling.assigned() || readMode == ReadMode::Scaled) + { + typeCtx.valueIn = descriptor.getSampleType(); + } + else + { + typeCtx.valueIn = postScaling.getInputSampleType(); + } + + { + typeCtx.valueLayout.rawSampleSize = descriptor.getRawSampleSize(); + auto dimensions = descriptor.getDimensions(); + if (dimensions.assigned() && dimensions.getCount() == 1) + { + typeCtx.valueLayout.valuesPerSample = dimensions[0].getSize(); + } + } + + if (typeCtx.valueOut == SampleType::Undefined) // Dynamically determine output type + { + typeCtx.valueOut = typeCtx.valueIn; + } + + bool valueTypesConvertible = TypedReadingUtils::isSampleTypeConvertible(typeCtx.valueIn, typeCtx.valueOut, false); + issues.set(QueueReaderIssue::ValueTypesNotConvertible, !valueTypesConvertible); +} + +void QueueReader::parseCachedDescriptors() +{ + parseDomainDescriptor(); + parseValueDescriptor(); +} + +size_t QueueReader::getNumberOfEventPacketsInQueue() +{ + size_t numberOfEventPackets = 0; + for (const auto& packet : packets) + { + if (packet.getType() == PacketType::Event) + ++numberOfEventPackets; + } + return numberOfEventPackets; +} + +bool QueueReader::dropUntilEvent() +{ + // Queue: d1 d2 E d3 -> E d3 + bool foundEvent = false; + size_t end = 0; + for (const auto& packet : packets) + { + if (packet.getType() == PacketType::Event) + { + foundEvent = true; + break; + } + ++end; + } + if (foundEvent) + { + packets.erase(packets.begin(), packets.begin() + end); + readingPosition = 0; + } + return foundEvent; +} + +END_NAMESPACE_OPENDAQ \ No newline at end of file diff --git a/core/opendaq/reader/src/signal_reader.cpp b/core/opendaq/reader/src/signal_reader.cpp index f7b9d50bc3..bb596e55d2 100644 --- a/core/opendaq/reader/src/signal_reader.cpp +++ b/core/opendaq/reader/src/signal_reader.cpp @@ -16,8 +16,6 @@ SignalReader::SignalReader(const InputPortConfigPtr& port, const LoggerComponentPtr& logger, bool globalIdFromSignal) : loggerComponent(logger) - , valueReader(createReaderForType(mode == ReadMode::RawValue ? SampleType::Undefined : valueReadType, nullptr)) - , domainReader(createReaderForType(domainReadType, nullptr)) , port(port) , connection(port.getConnection()) , readMode(mode) @@ -26,6 +24,10 @@ SignalReader::SignalReader(const InputPortConfigPtr& port, , commonSampleRate(-1) , globalIdFromSignal(globalIdFromSignal) { + trContext.domainIn = SampleType::Undefined; + trContext.domainOut = domainReadType; + trContext.valueIn = SampleType::Undefined; + trContext.valueOut = mode == ReadMode::RawValue ? SampleType::Undefined : valueReadType; } SignalReader::SignalReader(const SignalReader& old, @@ -33,9 +35,6 @@ SignalReader::SignalReader(const SignalReader& old, SampleType valueReadType, SampleType domainReadType) : loggerComponent(old.loggerComponent) - , valueReader(createReaderForType(old.readMode == ReadMode::RawValue ? SampleType::Undefined : valueReadType, - old.valueReader->getTransformFunction())) - , domainReader(createReaderForType(domainReadType, old.domainReader->getTransformFunction())) , port(old.port) , connection(port.getConnection()) , readMode(old.readMode) @@ -44,7 +43,17 @@ SignalReader::SignalReader(const SignalReader& old, , commonSampleRate(-1) , unused(old.unused) , globalIdFromSignal(old.globalIdFromSignal) + , trContext{} // In-out types might change { + // Preserve transform from the old reader + trContext.valueTransform = old.trContext.valueTransform; + trContext.domainTransform = old.trContext.domainTransform; + // Type-related state gets set as in new + trContext.domainIn = SampleType::Undefined; + trContext.domainOut = domainReadType; + trContext.valueIn = SampleType::Undefined; + trContext.valueOut = readMode == ReadMode::RawValue ? SampleType::Undefined : valueReadType; + info = old.info; port.setListener(listener); @@ -127,29 +136,29 @@ void SignalReader::handleDescriptorChanged(const EventPacketPtr& eventPacket) auto [valueDescriptorChanged, domainDescriptorChanged, newValueDescriptor, newDomainDescriptor] = parseDataDescriptorEventPacket(eventPacket); - if (valueDescriptorChanged && newValueDescriptor.assigned() && valueReader->getReadType() == SampleType::Undefined) + if (valueDescriptorChanged) { - SampleType valueType; - auto postScaling = newValueDescriptor.getPostScaling(); - if (!postScaling.assigned() || readMode == ReadMode::Scaled) - { - valueType = newValueDescriptor.getSampleType(); - } - else + if (newValueDescriptor.assigned() && trContext.valueOut == SampleType::Undefined) { - valueType = postScaling.getInputSampleType(); + SampleType valueType; + auto postScaling = newValueDescriptor.getPostScaling(); + if (!postScaling.assigned() || readMode == ReadMode::Scaled) + { + valueType = newValueDescriptor.getSampleType(); + } + else + { + valueType = postScaling.getInputSampleType(); + } + + trContext.valueOut = valueType; } - valueReader = createReaderForType(valueType, valueReader->getTransformFunction()); - } - - if (valueDescriptorChanged) - { - invalid = !valueReader->handleDescriptorChanged(newValueDescriptor, readMode); + invalid = !onValueDescriptorUpdate(newValueDescriptor); } if (domainDescriptorChanged) { - auto validDomain = domainReader->handleDescriptorChanged(newDomainDescriptor, readMode); + auto validDomain = onDomainDescriptorUpdate(newDomainDescriptor); if (validDomain && newDomainDescriptor.assigned()) { auto newResolution = newDomainDescriptor.getTickResolution(); @@ -219,7 +228,7 @@ void SignalReader::setStartInfo(std::chrono::system_clock::time_point minEpoch, synced = SyncStatus::Unsynchronized; } -std::unique_ptr SignalReader::readStartDomain() +std::unique_ptr SignalReader::readDomainStart() { DataPacketPtr domainPacket = info.dataPacket.getDomainPacket(); if (!domainPacket.assigned()) @@ -227,14 +236,42 @@ std::unique_ptr SignalReader::readStartDomain() DAQ_THROW_EXCEPTION(InvalidStateException, "Packet must have a domain packet assigned!"); } - if (domainPacket.getDataDescriptor().getRule().getType() == DataRuleType::Linear) - { - return domainReader->readStartLinear(domainPacket, info.prevSampleIndex, domainInfo); - } - else - { - return domainReader->readStart(domainPacket.getData(), info.prevSampleIndex, domainInfo); - } + return TypedReadingUtils::readDomainValue(trContext.domainIn, + trContext.domainOut, + trContext.domainLayout, + domainPacket, + info.prevSampleIndex, + trContext.domainInfo); +} + +const DomainInfo& SignalReader::getDomainInfo() const +{ + return trContext.domainInfo; +} + +SampleType SignalReader::getValueReadType() const +{ + return trContext.valueOut; +} + +void SignalReader::setValueTransformFunction(FunctionPtr transform) +{ + trContext.valueTransform = std::move(transform); +} + +void SignalReader::setDomainTransformFunction(FunctionPtr transform) +{ + trContext.domainTransform = std::move(transform); +} + +FunctionPtr SignalReader::getValueTransformFunction() const +{ + return trContext.valueTransform; +} + +FunctionPtr SignalReader::getDomainTransformFunction() const +{ + return trContext.domainTransform; } bool SignalReader::isFirstPacketEvent() @@ -389,7 +426,7 @@ bool SignalReader::skipUntilLastEventPacket() return hasEventPacket; } -bool SignalReader::sync(const Comparable& commonStart, std::chrono::system_clock::rep* firstSampleAbsoluteTimestamp) +bool SignalReader::sync(const DomainValue* commonStart, std::chrono::system_clock::rep* firstSampleAbsoluteTimestamp) { if (synced == SyncStatus::Synchronized) { @@ -411,15 +448,12 @@ bool SignalReader::sync(const Comparable& commonStart, std::chrono::system_clock { auto domainPacket = info.dataPacket.getDomainPacket(); // Check if commonStart can be reached within the current packet - if (domainPacket.getDataDescriptor().getRule().getType() == DataRuleType::Linear) - { - info.prevSampleIndex = domainReader->getOffsetToLinear(domainInfo, commonStart, domainPacket, &cachedFirstTimestamp); - } - else - { - info.prevSampleIndex = domainReader->getOffsetTo( - domainInfo, commonStart, domainPacket.getData(), domainPacket.getSampleCount(), &cachedFirstTimestamp); - } + info.prevSampleIndex = TypedReadingUtils::findDomainValue(trContext.domainIn, + trContext.domainOut, + trContext.domainLayout, + domainPacket, + commonStart, + &cachedFirstTimestamp); if (info.prevSampleIndex == static_cast(-1)) { @@ -567,7 +601,15 @@ ErrCode SignalReader::readPacketData() if (info.values != nullptr) { - ErrCode errCode = valueReader->readData(getValuePacketData(info.dataPacket), info.prevSampleIndex, &info.values, toRead); + ErrCode errCode = TypedReadingUtils::readData(trContext.valueIn, + trContext.valueOut, + false, + trContext.valueLayout, + getValuePacketData(info.dataPacket), + info.prevSampleIndex, + &info.values, + toRead, + trContext.valueTransform); OPENDAQ_RETURN_IF_FAILED(errCode); } @@ -582,13 +624,30 @@ ErrCode SignalReader::readPacketData() LOG_T("[Reading: {} ", port.getSignal().getLocalId()); auto domainPacket = dataPacket.getDomainPacket(); - ErrCode errCode = domainReader->readData(domainPacket.getData(), info.prevSampleIndex, &info.domainValues, toRead); + ErrCode errCode = TypedReadingUtils::readData(trContext.domainIn, + trContext.domainOut, + true, + trContext.domainLayout, + domainPacket.getData(), + info.prevSampleIndex, + &info.domainValues, + toRead, + trContext.domainTransform); + if (errCode == OPENDAQ_ERR_INVALIDSTATE) { if (!trySetDomainSampleType(domainPacket)) return DAQ_EXTEND_ERROR_INFO(errCode, "Failed to set domain sample type for packet"); daqClearErrorInfo(); - errCode = domainReader->readData(domainPacket.getData(), info.prevSampleIndex, &info.domainValues, toRead); + errCode = TypedReadingUtils::readData(trContext.domainIn, + trContext.domainOut, + true, + trContext.domainLayout, + domainPacket.getData(), + info.prevSampleIndex, + &info.domainValues, + toRead, + trContext.domainTransform); } LOG_T("]"); @@ -609,18 +668,75 @@ ErrCode SignalReader::readPacketData() return OPENDAQ_SUCCESS; } -bool SignalReader::trySetDomainSampleType(const daq::DataPacketPtr& domainPacket) const +bool SignalReader::trySetDomainSampleType(const daq::DataPacketPtr& domainPacket) { ObjectPtr errorInfo; daqGetErrorInfo(&errorInfo); daqClearErrorInfo(); auto dataDescriptor = domainPacket.getDataDescriptor(); - if (domainReader->handleDescriptorChanged(dataDescriptor, readMode)) + if (onDomainDescriptorUpdate(dataDescriptor)) return true; daqSetErrorInfo(errorInfo); return false; } +bool SignalReader::onValueDescriptorUpdate(const DataDescriptorPtr& valueDescriptor) +{ + if (!valueDescriptor.assigned()) + return false; + + const auto postScaling = valueDescriptor.getPostScaling(); + if (!postScaling.assigned() || readMode == ReadMode::Scaled) + { + trContext.valueIn = valueDescriptor.getSampleType(); + } + else + { + trContext.valueIn = postScaling.getInputSampleType(); + } + + trContext.valueLayout.rawSampleSize = valueDescriptor.getRawSampleSize(); + auto dimensions = valueDescriptor.getDimensions(); + if (dimensions.assigned() && dimensions.getCount() == 1) + { + trContext.valueLayout.valuesPerSample = dimensions[0].getSize(); + } + + trContext.valueLayout.descriptor = valueDescriptor; + + return TypedReadingUtils::isSampleTypeConvertible(trContext.valueIn, trContext.valueOut, false); +} + +bool SignalReader::onDomainDescriptorUpdate(const DataDescriptorPtr& domainDescriptor) +{ + if (!domainDescriptor.assigned()) + return false; + + const auto postScaling = domainDescriptor.getPostScaling(); + if (!postScaling.assigned() || readMode == ReadMode::Scaled) + { + trContext.domainIn = domainDescriptor.getSampleType(); + } + else + { + trContext.domainIn = postScaling.getInputSampleType(); + } + + trContext.domainLayout.rawSampleSize = domainDescriptor.getRawSampleSize(); + auto dimensions = domainDescriptor.getDimensions(); + if (dimensions.assigned() && dimensions.getCount() == 1) + { + trContext.domainLayout.valuesPerSample = dimensions[0].getSize(); + } + + trContext.domainLayout.descriptor = domainDescriptor; + + trContext.domainInfo = DomainInfo::fromDescriptor(domainDescriptor); + + return TypedReadingUtils::isSampleTypeConvertible(trContext.domainIn, trContext.domainOut, true); +} + + END_NAMESPACE_OPENDAQ diff --git a/core/opendaq/reader/src/typed_reading_utils.cpp b/core/opendaq/reader/src/typed_reading_utils.cpp new file mode 100644 index 0000000000..7afc8f41f6 --- /dev/null +++ b/core/opendaq/reader/src/typed_reading_utils.cpp @@ -0,0 +1,671 @@ +#include + +#include +#include +#include +#include + +BEGIN_NAMESPACE_OPENDAQ + +namespace +{ + +template +struct TypeTag +{ + using Type = T; +}; + +template +decltype(auto) visitSampleType(SampleType sampleType, Visitor&& visitor) +{ + switch (sampleType) + { + case SampleType::Float32: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Float64: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::UInt8: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Int8: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::UInt16: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Int16: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::UInt32: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Int32: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::UInt64: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Int64: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::RangeInt64: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::ComplexFloat32: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::ComplexFloat64: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Struct: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Undefined: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Binary: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::String: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::Null: + return std::forward(visitor)(TypeTag::Type>{}); + case SampleType::_count: + return std::forward(visitor)(TypeTag::Type>{}); + } + DAQ_THROW_EXCEPTION(NotSupportedException, "The requested sample-type is unsupported or invalid."); +} + +std::string_view format_as(SampleType sampleType); + +bool validateInputType(SampleType sampleType, bool isDomain) +{ + switch (sampleType) + { + case SampleType::Float32: + case SampleType::Float64: + case SampleType::UInt8: + case SampleType::Int8: + case SampleType::UInt16: + case SampleType::Int16: + case SampleType::UInt32: + case SampleType::Int32: + case SampleType::UInt64: + case SampleType::Int64: + return true; + case SampleType::RangeInt64: + return true; // TODO: Not sure what kind of support we actually have + case SampleType::ComplexFloat32: + case SampleType::ComplexFloat64: + case SampleType::Struct: + case SampleType::Undefined: + case SampleType::Binary: + case SampleType::String: + if (isDomain) + DAQ_THROW_EXCEPTION(NotSupportedException, "Using the SampleType {} as a domain is not supported", format_as(sampleType)); + return true; + case SampleType::Null: + DAQ_THROW_EXCEPTION(NotSupportedException, "Null input sample type encountered."); + case SampleType::_count: + default: + DAQ_THROW_EXCEPTION(InvalidStateException, "Unexpected input sample type encountered."); + } +} + +bool validateOutputType(SampleType sampleType, bool isDomain) +{ + switch (sampleType) + { + case SampleType::Float32: + case SampleType::Float64: + case SampleType::UInt8: + case SampleType::Int8: + case SampleType::UInt16: + case SampleType::Int16: + case SampleType::UInt32: + case SampleType::Int32: + case SampleType::UInt64: + case SampleType::Int64: + case SampleType::RangeInt64: + return true; + case SampleType::ComplexFloat32: + case SampleType::ComplexFloat64: + case SampleType::Struct: + case SampleType::Undefined: + return !isDomain; + case SampleType::Binary: + case SampleType::String: + case SampleType::Null: + case SampleType::_count: + default: + DAQ_THROW_EXCEPTION(NotSupportedException, "The requested output sample type is unsupported or invalid."); + } +} + +bool validateInputOutputType(SampleType inputType, SampleType outputType, bool isDomain) +{ + // TODO: Implement combination checking (float -> int for domains etc.) + return true; +} + +template +decltype(auto) visitTwoSampleTypes(SampleType inputType, SampleType outputType, bool isDomain, Visitor&& visitor) +{ + if (!validateInputType(inputType, isDomain)) + DAQ_THROW_EXCEPTION(NotSupportedException, "Input sample type not supported."); + + if (!validateOutputType(outputType, isDomain)) + DAQ_THROW_EXCEPTION(NotSupportedException, "Output sample type not supported."); + + if (!validateInputOutputType(inputType, outputType, isDomain)) + DAQ_THROW_EXCEPTION(NotSupportedException, "Output sample type not supported."); + + return visitSampleType(inputType, + [&](auto inputTag) -> decltype(auto) + { + return visitSampleType(outputType, + [&](auto outputTag) -> decltype(auto) + { return std::forward(visitor)(inputTag, outputTag); }); + }); +} + +std::string_view format_as(SampleType sampleType) +{ + switch (sampleType) + { + case SampleType::Float32: + return "Float32"; + case SampleType::Float64: + return "Float64"; + case SampleType::UInt8: + return "UInt8"; + case SampleType::Int8: + return "Int8"; + case SampleType::UInt16: + return "UInt16"; + case SampleType::Int16: + return "Int16"; + case SampleType::UInt32: + return "UInt32"; + case SampleType::Int32: + return "Int32"; + case SampleType::UInt64: + return "UInt64"; + case SampleType::Int64: + return "Int64"; + case SampleType::RangeInt64: + return "RangeInt64"; + case SampleType::ComplexFloat32: + return "ComplexFloat32"; + case SampleType::ComplexFloat64: + return "ComplexFloat64"; + case SampleType::Struct: + return "Struct"; + case SampleType::Undefined: + return "Undefined"; + case SampleType::Binary: + return "Binary"; + case SampleType::String: + return "String"; + case SampleType::Null: + return "Null"; + case SampleType::_count: + return "Count"; + } + return "Unknown"; +} + +} + +namespace detail +{ + +template +bool isSampleTypeConvertible(bool isDomain) +{ + if constexpr (std::is_same_v) + { + return !isDomain; + } + else + { + return std::is_convertible_v; + } +} + +template +ErrCode readData(const ReadLayout& readLayout, + void* inputBuffer, + SizeT offset, + void** outputBuffer, + SizeT toRead, + const FunctionPtr& transformFunction = nullptr) +{ + OPENDAQ_PARAM_NOT_NULL(inputBuffer); + OPENDAQ_PARAM_NOT_NULL(outputBuffer); + OPENDAQ_PARAM_NOT_NULL(readLayout.descriptor.getObject()); + + const auto& rawSampleSize = readLayout.rawSampleSize; + const auto& valuesPerSample = readLayout.valuesPerSample; + const auto& dataDescriptor = readLayout.descriptor; + + if constexpr (std::is_same_v) + { + if (transformFunction.assigned()) + return DAQ_MAKE_ERROR_INFO(OPENDAQ_ERR_NOT_SUPPORTED, "Transform function for void reader not supported."); + + const auto dataStart = static_cast(static_cast(inputBuffer) + offset * rawSampleSize); + const auto toReadInBytes = rawSampleSize * toRead; + const auto dataOut = *outputBuffer; + std::memcpy(dataOut, dataStart, toReadInBytes); + *outputBuffer = static_cast(static_cast(dataOut) + toReadInBytes); + return OPENDAQ_SUCCESS; + } + else if constexpr (std::is_convertible_v) + { + auto* dataStart = static_cast(inputBuffer) + (offset * valuesPerSample); + auto* dataOut = static_cast(*outputBuffer); + + if (transformFunction.assigned()) + { + transformFunction.call((Int) dataStart, (Int) dataOut, toRead, dataDescriptor); + + *outputBuffer = dataOut + (valuesPerSample * toRead); + return OPENDAQ_SUCCESS; + } + + // If the type of samples is the same, then just copy + if constexpr (std::is_same_v) + { + // Returns the pointer to the value after the last copied one + *outputBuffer = std::copy_n(dataStart, valuesPerSample * toRead, dataOut); // C4244 - possible data loss due to conversion + } + else + { + for (std::size_t i = 0; i < toRead * valuesPerSample; ++i) + { + dataOut[i] = static_cast(dataStart[i]); // C4244 - possible data loss due to conversion + } + + // Set the pointer to the value after the last copied one + *outputBuffer = &dataOut[toRead]; + } + + return OPENDAQ_SUCCESS; + } + else + { + return DAQ_MAKE_ERROR_INFO(OPENDAQ_ERR_NOT_SUPPORTED, + "Implicit conversion from packet data-type to the read data-type is not supported."); + } +} + +template +void getDeltaStart(const daq::DictPtr& params, OutputT& delta, OutputT& start) +{ + if constexpr (std::is_same_v) + { + DAQ_THROW_EXCEPTION(NotSupportedException, "Void reader should not be used for domain."); + } + if constexpr (!std::is_convertible_v) + { + DAQ_THROW_EXCEPTION(NotSupportedException, "Implicit conversion from packet data-type to the read data-type is not supported."); + } + else + { + delta = static_cast(params.get("delta")); + start = static_cast(params.get("start")); + } +} + +template +std::unique_ptr readDomainValueLinear(const DataPacketPtr& domainPacket, SizeT index, const DomainInfo& domainInfo) +{ + if constexpr (std::is_same_v || !std::is_integral_v) + { + DAQ_THROW_EXCEPTION(NotSupportedException, + "ReadDomainValueLinear not supported for the selected output type (void / non-integral)."); + return {}; + } + else + { + const DataRulePtr dataRule = domainPacket.getDataDescriptor().getRule(); + NumberPtr packetOffset = domainPacket.getOffset(); + const auto parameters = dataRule.getParameters(); + OutputT delta, start; + getDeltaStart(parameters, delta, start); + + int64_t rdOffset = 0; + auto refDomainInfo = domainPacket.getDataDescriptor().getReferenceDomainInfo(); + if (refDomainInfo.assigned()) + { + const IntPtr referenceDomainOffset = refDomainInfo.getReferenceDomainOffset(); + if (referenceDomainOffset.assigned()) + { + rdOffset = referenceDomainOffset; + } + } + + OutputT timestamp = + start + static_cast(rdOffset) + static_cast(packetOffset.getIntValue()) + delta * static_cast(index); + return std::make_unique>(domainInfo, timestamp); + } +} + +template +std::unique_ptr readDomainValue(const ReadLayout& readLayout, + const DataPacketPtr& domainPacket, + SizeT index, + const DomainInfo& domainInfo) +{ + if constexpr (std::is_same_v) + { + DAQ_THROW_EXCEPTION(NotSupportedException, "ReadDomainValueLinear not supported for the void output type."); + return {}; + } + else + { + auto descriptor = domainPacket.getDataDescriptor(); + if (!descriptor.assigned()) + DAQ_THROW_EXCEPTION(InvalidStateException, "Packet should have descriptor assigned."); + + OutputT timestamp{}; + void* data = ×tamp; + readData(readLayout, domainPacket.getData(), index, &data, 1, nullptr); + return std::make_unique>(domainInfo, timestamp); + } +} + +template +SizeT findDomainValueLinear(const DataPacketPtr& domainPacket, + const DomainValue* target, + [[maybe_unused]] std::chrono::system_clock::rep* absoluteTimestamp) +{ + if constexpr (!std::is_same_v && std::is_integral_v && std::is_convertible_v) + { + const SizeT sampleCount = domainPacket.getSampleCount(); + if (sampleCount == 0) + { + return static_cast(-1); + } + + const DataRulePtr& dataRule = domainPacket.getDataDescriptor().getRule(); + const auto parameters = dataRule.getParameters(); + + int64_t rdOffset = 0; + { // Extract reference domain offset + auto refDomainInfo = domainPacket.getDataDescriptor().getReferenceDomainInfo(); + if (refDomainInfo.assigned()) + { + const IntPtr referenceDomainOffset = refDomainInfo.getReferenceDomainOffset(); + if (referenceDomainOffset.assigned()) + { + rdOffset = referenceDomainOffset; + } + } + } + + OutputT ruleDelta, ruleStart; + getDeltaStart(parameters, ruleDelta, ruleStart); + NumberPtr packetOffset = domainPacket.getOffset(); + + // Total packet offset in signal resolution ticks. + const OutputT startTick = ruleStart + static_cast(rdOffset) + static_cast(packetOffset.getIntValue()); + const OutputT previousEndTick = startTick - ruleDelta; + + // Tick of the last sample in signal resolution ticks + const SizeT packetSize = domainPacket.getSampleCount(); + OutputT endTick = startTick + ruleDelta * static_cast(packetSize - 1); + + const DomainValueImpl* typedTarget = dynamic_cast*>(target); + OutputT targetValue = typedTarget->getValue(); + + /* + The function returns the index k of the first tick where the tick_k >= target + (in max resolution ticks since min epoch). The k-th sample is the answer for any target inside + interval (tick_{k-1}, tick_k]. The n-th packet then covers ticks (endTick_{n-1}, endTick]. + Using the linear rule, the last sample in the previous packet was at endTick_{n-1} = startTick_n - ruleDelta. + + Example: Packet with ticks [12, 15, 18] will return index 0 for targets [10, 11, 12] and -1 for <=9 and >=19. + */ + if (targetValue <= previousEndTick || targetValue > endTick) + { + // Target is outside this packet + return static_cast(-1); + } + + // Ticks from the end of the previous packet's end. + SizeT ticksToTarget = targetValue - previousEndTick; + // Minus one due to calculation from the previous packet end tick. + SizeT index = static_cast((ticksToTarget + ruleDelta - 1) / ruleDelta) - 1; + + assert(index < packetSize && "Index out of bounds"); + + if (absoluteTimestamp) + { + if constexpr (IsTemplateOf::value || IsTemplateOf::value) + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + else + { + // Tick corresponding to index in signal's resolution ticks. + OutputT tick = startTick + static_cast(index) * ruleDelta; + auto readValueSysTime = reader::toSysTime(tick, target->getDomain().epoch, target->getDomain().resolution); + *absoluteTimestamp = readValueSysTime.time_since_epoch().count(); + } + } + return index; + } + else + { + DAQ_THROW_EXCEPTION(NotSupportedException, "Implicit conversion from packet data-type to the read data-type is not supported."); + } +} + +template +SizeT findDomainValue(const ReadLayout& readLayout, + const DataPacketPtr& domainPacket, + const DomainValue* target, + [[maybe_unused]] std::chrono::system_clock::rep* absoluteTimestamp) +{ + void* inputBuffer = domainPacket.getData(); + SizeT size = domainPacket.getSampleCount(); + + if (!inputBuffer) + DAQ_THROW_EXCEPTION(ArgumentNullException, "Packet with null data buffer"); + + if constexpr (std::is_convertible_v && !std::is_same_v && !std::is_same_v) + { + InputT* domainBuffer = static_cast(inputBuffer); + + const DomainValueImpl* typedTarget = dynamic_cast*>(target); + OutputT targetValue = typedTarget->getValue(); + + for (std::size_t i = 0; i < size * readLayout.valuesPerSample; ++i) + { + OutputT value = static_cast(domainBuffer[i]); // C4244 - possible data loss due to conversion + + bool greaterEqual = false; + if constexpr (IsTemplateOf::value) + { + if (value.start >= targetValue.start) + { + if (absoluteTimestamp) + { + auto readValueSysTime = reader::toSysTime(value.start, target->getDomain().epoch, target->getDomain().resolution); + *absoluteTimestamp = readValueSysTime.time_since_epoch().count(); + } + greaterEqual = true; + } + } + else if constexpr (!IsTemplateOf::value) + { + if (value >= targetValue) + { + if (absoluteTimestamp) + { + auto readValueSysTime = reader::toSysTime(value, target->getDomain().epoch, target->getDomain().resolution); + *absoluteTimestamp = readValueSysTime.time_since_epoch().count(); + } + greaterEqual = true; + } + } + else + { + DAQ_THROW_EXCEPTION(NotSupportedException); + } + + if (greaterEqual) + { + return i / readLayout.valuesPerSample; + } + } + + return static_cast(-1); + } + else + { + DAQ_THROW_EXCEPTION(NotSupportedException, "Implicit conversion from packet data-type to the read data-type is not supported."); + } +} + +} + +ReadLayout TypedReadingUtils::createReadLayout(const DataDescriptorPtr& descriptor) +{ + if (!descriptor.assigned()) + DAQ_THROW_EXCEPTION(ArgumentNullException, "Descriptor must be assigned!"); + + const SizeT rawSampleSize = descriptor.getRawSampleSize(); + SizeT valuesPerSample = 1; + auto dimensions = descriptor.getDimensions(); + if (dimensions.assigned() && dimensions.getCount() == 1) + { + valuesPerSample = dimensions[0].getSize(); + } + + return {descriptor, rawSampleSize, valuesPerSample}; +} + +bool TypedReadingUtils::isSampleTypeConvertible(SampleType in, SampleType out, bool isDomain) +{ + // TODO: Detais about limiting allowed types (not throwing unless necessary) + switch (in) + { + case SampleType::Struct: + case SampleType::Invalid: + case SampleType::Null: + case SampleType::_count: + return false; + default: + break; + } + + if (isDomain) + { + switch (in) + { + case SampleType::Float32: + return in == SampleType::Float32 || in == SampleType::Float64; + case SampleType::Float64: + return in == SampleType::Float32 || in == SampleType::Float64; + case SampleType::ComplexFloat32: + case SampleType::ComplexFloat64: + case SampleType::Binary: + return false; + default: + break; + } + } + + return visitTwoSampleTypes(in, + out, + isDomain, + [&](auto inputTag, auto outputTag) -> bool + { + using InputT = typename decltype(inputTag)::Type; + using OutputT = typename decltype(outputTag)::Type; + return detail::isSampleTypeConvertible(isDomain); + }); +} + +std::unique_ptr TypedReadingUtils::readDomainValue(SampleType in, + SampleType out, + const ReadLayout& readLayout, + const DataPacketPtr& domainPacket, + SizeT index, + const DomainInfo& domainInfo) +{ + const DataRulePtr dataRule = domainPacket.getDataDescriptor().getRule(); + if (dataRule.getType() == DataRuleType::Linear) + { + return visitTwoSampleTypes(in, + out, + true, + [&](auto inputTag, auto outputTag) -> std::unique_ptr + { + using InputT = typename decltype(inputTag)::Type; + using OutputT = typename decltype(outputTag)::Type; + return detail::readDomainValueLinear(domainPacket, index, domainInfo); + }); + } + else + { + return visitTwoSampleTypes(in, + out, + true, + [&](auto inputTag, auto outputTag) -> std::unique_ptr + { + using InputT = typename decltype(inputTag)::Type; + using OutputT = typename decltype(outputTag)::Type; + return detail::readDomainValue(readLayout, domainPacket, index, domainInfo); + }); + } +} + +SizeT TypedReadingUtils::findDomainValue(SampleType in, + SampleType out, + const ReadLayout& readLayout, + const DataPacketPtr& domainPacket, + const DomainValue* target, + std::chrono::system_clock::rep* firstSampleAbsoluteTime) +{ + const DataRulePtr dataRule = domainPacket.getDataDescriptor().getRule(); + if (dataRule.getType() == DataRuleType::Linear) + { + return visitTwoSampleTypes(in, + out, + true, + [&](auto inputTag, auto outputTag) -> SizeT + { + using InputT = typename decltype(inputTag)::Type; + using OutputT = typename decltype(outputTag)::Type; + return detail::findDomainValueLinear(domainPacket, target, firstSampleAbsoluteTime); + }); + } + else + { + return visitTwoSampleTypes(in, + out, + true, + [&](auto inputTag, auto outputTag) -> SizeT + { + using InputT = typename decltype(inputTag)::Type; + using OutputT = typename decltype(outputTag)::Type; + return detail::findDomainValue( + readLayout, domainPacket, target, firstSampleAbsoluteTime); + }); + } +} + +ErrCode TypedReadingUtils::readData(SampleType in, + SampleType out, + bool isDomain, + const ReadLayout& readLayout, + void* inputBuffer, + SizeT offset, + void** outputBuffer, + SizeT count, + const FunctionPtr transform) +{ + return visitTwoSampleTypes(in, + out, + isDomain, + [&](auto inputTag, auto outputTag) -> ErrCode + { + using InputT = typename decltype(inputTag)::Type; + using OutputT = typename decltype(outputTag)::Type; + return detail::readData( + readLayout, inputBuffer, offset, outputBuffer, count, transform); + }); +} + +END_NAMESPACE_OPENDAQ \ No newline at end of file diff --git a/core/opendaq/reader/tests/CMakeLists.txt b/core/opendaq/reader/tests/CMakeLists.txt index 25065d536b..a4f97badc5 100644 --- a/core/opendaq/reader/tests/CMakeLists.txt +++ b/core/opendaq/reader/tests/CMakeLists.txt @@ -13,6 +13,11 @@ set(TEST_SOURCES test_factories.cpp test_time_reader.cpp test_multi_reader.cpp test_stream_reader_from_input_port.cpp + test_domain_value.cpp + test_typed_reading.cpp + test_queue_reader.cpp + ../src/typed_reading_utils.cpp + ../src/queue_reader.cpp ) opendaq_prepare_test_runner(TEST_APP FOR ${MODULE_NAME} @@ -32,14 +37,15 @@ if (MSVC) target_compile_options(${TEST_APP} PRIVATE /bigobj) endif() -target_link_libraries(${TEST_APP} PRIVATE daq::coreobjects -) add_test(NAME ${TEST_APP} COMMAND $ WORKING_DIRECTORY $ ) +target_link_libraries(${TEST_APP} PRIVATE daq::coreobjects daq::opendaq +) + if (OPENDAQ_ENABLE_COVERAGE) setup_target_for_coverage(${TEST_APP}coverage ${TEST_APP} ${TEST_APP}coverage) endif() diff --git a/core/opendaq/reader/tests/test_domain_value.cpp b/core/opendaq/reader/tests/test_domain_value.cpp new file mode 100644 index 0000000000..41f27a47fe --- /dev/null +++ b/core/opendaq/reader/tests/test_domain_value.cpp @@ -0,0 +1,277 @@ +#include + +#include + +#include +#include + +#include + +using DomainValueTest = testing::Test; + +TEST_F(DomainValueTest, DomainInfoComparison) +{ + daq::DomainInfo info1 = {daq::reader::parseEpoch("1970-01-01T00:00:00+00:00"), daq::Ratio(1, 1000)}; + daq::DomainInfo info2 = {daq::reader::parseEpoch("1970-01-01T00:00:00+00:00"), daq::Ratio(1, 1000)}; + ASSERT_TRUE(info1 == info2); + + // Different epoch + daq::DomainInfo info3 = {daq::reader::parseEpoch("1999-01-01T00:00:00+00:00"), daq::Ratio(1, 1000)}; + ASSERT_FALSE(info3 == info1); + + // Different resolution denominator + daq::DomainInfo info4 = {daq::reader::parseEpoch("1970-01-01T00:00:00+00:00"), daq::Ratio(1, 2000)}; + ASSERT_FALSE(info4 == info1); + + // Different resolution numerator + daq::DomainInfo info5 = {daq::reader::parseEpoch("1970-01-01T00:00:00+00:00"), daq::Ratio(4, 1000)}; + ASSERT_FALSE(info5 == info1); + + // Different everything + daq::DomainInfo info6 = {daq::reader::parseEpoch("1999-01-01T00:00:00+00:00"), daq::Ratio(5, 2000)}; + ASSERT_FALSE(info6 == info1); + + // Unassigned resolution + daq::DomainInfo info7 = {daq::reader::parseEpoch("1999-01-01T00:00:00+00:00"), nullptr}; + ASSERT_THROW((void) (info7 == info1), daq::InvalidParameterException); +} + +template +void checkTypedDomainValue() +{ + auto epoch = daq::reader::parseEpoch("2026-01-01T00:00:00+00:00"); + daq::DomainInfo domain = {epoch, daq::Ratio(1, 1000000)}; + + T tick; + if constexpr (std::is_same_v) + { + tick = {12345, 12350}; + } + else + { + tick = 12345; + } + std::unique_ptr value = std::make_unique>(domain, tick); + ASSERT_TRUE(value->getDomain() == domain); + + auto* castValue = dynamic_cast*>(value.get()); + ASSERT_TRUE(castValue != nullptr); + + ASSERT_EQ(castValue->getValue(), tick); + + T greaterTick; + if constexpr (std::is_same_v) + { + greaterTick = {12351, 12360}; + } + else + { + greaterTick = 12351; + } + std::unique_ptr greaterValue = std::make_unique>(domain, greaterTick); + ASSERT_TRUE(*value < *greaterValue); + ASSERT_FALSE(*greaterValue < *value); + ASSERT_FALSE(*value < *value); +} + +TEST_F(DomainValueTest, StoresDaqInt) +{ + checkTypedDomainValue(); +} + +TEST_F(DomainValueTest, StoresDaqUInt) +{ + checkTypedDomainValue(); +} + +TEST_F(DomainValueTest, StoresDaqFloat) +{ + checkTypedDomainValue(); +} + +TEST_F(DomainValueTest, StoresDaqRange) +{ + checkTypedDomainValue(); +} + +template +void checkTypedDomainValueComplex() +{ + auto epoch = daq::reader::parseEpoch("2026-01-01T00:00:00+00:00"); + daq::DomainInfo domain = {epoch, daq::Ratio(1, 1000000)}; + + T tick = {12.3, 15.4}; + std::unique_ptr value = std::make_unique>(domain, tick); + ASSERT_TRUE(value->getDomain() == domain); + + auto* castValue = dynamic_cast*>(value.get()); + ASSERT_TRUE(castValue != nullptr); + + ASSERT_THROW(castValue->getValue(), daq::NotSupportedException); + + T greaterTick = {16.2, 25.3}; + + std::unique_ptr greaterValue = std::make_unique>(domain, greaterTick); + ASSERT_THROW((void) (*value < *greaterValue), daq::NotSupportedException); + ASSERT_THROW((void) (*greaterValue < *value), daq::NotSupportedException); +} + +TEST_F(DomainValueTest, ThrowsForComplex) +{ + checkTypedDomainValueComplex(); + checkTypedDomainValueComplex(); +} + +TEST_F(DomainValueTest, Scaling) +{ + std::string commonEpochString = "2026-01-01T00:00:00+00:00"; + std::chrono::system_clock::time_point commonEpoch = daq::reader::parseEpoch(commonEpochString); + daq::RatioPtr commonResolution = daq::Ratio(1, 1000000); + + daq::DomainInfo commonDomain = {commonEpoch, commonResolution}; + + daq::RatioPtr resolution1 = daq::Ratio(1, 5000); + daq::DomainInfo domain1 = {commonEpoch, resolution1}; + + auto value1 = std::make_unique>(domain1, 10000); + ASSERT_EQ(value1->getValue(), 10000u); + + auto value1InCommonDomain = value1->toCommonDomain(commonDomain); + auto* value1InCommonDomainP = dynamic_cast*>(value1InCommonDomain.get()); + ASSERT_EQ(value1InCommonDomainP->getValue(), 2000000u); + + auto value1BackInRegularDomain = value1InCommonDomain->fromCommonDomain(domain1); + auto* value1BackInRegularDomainP = dynamic_cast*>(value1BackInRegularDomain.get()); + ASSERT_EQ(value1BackInRegularDomainP->getValue(), 10000u); +} + +TEST_F(DomainValueTest, Offset) +{ + std::string commonEpochString = "1970-01-01T00:00:00+00:00"; + std::chrono::system_clock::time_point commonEpoch = daq::reader::parseEpoch(commonEpochString); + daq::RatioPtr commonResolution = daq::Ratio(1, 1000); + + daq::DomainInfo commonDomain = {commonEpoch, commonResolution}; + + std::string epochString1 = "1970-01-01T00:00:50+00:00"; + std::chrono::system_clock::time_point epoch1 = daq::reader::parseEpoch(epochString1); + daq::RatioPtr resolution1 = daq::Ratio(1, 1000); + daq::DomainInfo domain1 = {epoch1, resolution1}; + + auto value1 = std::make_unique>(domain1, 13000); + ASSERT_EQ(value1->getValue(), 13000u); + + auto value1InCommonDomain = value1->toCommonDomain(commonDomain); + auto* value1InCommonDomainP = dynamic_cast*>(value1InCommonDomain.get()); + ASSERT_EQ(value1InCommonDomainP->getValue(), 63000u); + ASSERT_EQ(value1InCommonDomainP->getDomain(), commonDomain); + + auto value1BackInRegularDomain = value1InCommonDomain->fromCommonDomain(domain1); + auto* value1BackInRegularDomainP = dynamic_cast*>(value1BackInRegularDomain.get()); + ASSERT_EQ(value1BackInRegularDomainP->getValue(), 13000u); + ASSERT_EQ(value1BackInRegularDomainP->getDomain(), domain1); +} + +TEST_F(DomainValueTest, SameDomainSameType) +{ + std::string commonEpochString = "1970-01-01T00:00:00+00:00"; + std::chrono::system_clock::time_point commonEpoch = daq::reader::parseEpoch(commonEpochString); + daq::RatioPtr commonResolution = daq::Ratio(1, 1000); + + daq::DomainInfo commonDomain = {commonEpoch, commonResolution}; + + std::string epochString1 = "1970-01-01T00:00:50+00:00"; + std::chrono::system_clock::time_point epoch1 = daq::reader::parseEpoch(epochString1); + daq::RatioPtr resolution1 = daq::Ratio(1, 1000); + daq::DomainInfo domain1 = {epoch1, resolution1}; + + auto value1 = std::make_unique>(domain1, 13000); + daq::DomainValue* value1P = value1.get(); + auto value1InCommonDomain = value1->toCommonDomain(commonDomain); + auto* value1InCommonDomainP = dynamic_cast*>(value1InCommonDomain.get()); + + ASSERT_THROW((void) (*value1P < *value1InCommonDomainP), daq::InvalidParameterException); + + auto value2 = std::make_unique>(domain1, 1213000); + daq::DomainValue* value2P = value2.get(); + + ASSERT_THROW((void) (*value1 < *value2), daq::InvalidParameterException); +} + +TEST_F(DomainValueTest, RealisticTimeStamp) +{ + std::string commonEpochString = "1970-01-01T00:00:00+00:00"; + std::chrono::system_clock::time_point commonEpoch = daq::reader::parseEpoch(commonEpochString); + daq::RatioPtr commonResolution = daq::Ratio(1, 1000000000); + daq::DomainInfo commonDomain = {commonEpoch, commonResolution}; + + std::string epochString1 = "2026-05-27T00:00:01+00:00"; + std::chrono::system_clock::time_point epoch1 = daq::reader::parseEpoch(epochString1); + daq::RatioPtr resolution1 = daq::Ratio(1, 1000000); + daq::DomainInfo domain1 = {epoch1, resolution1}; + + auto value1 = std::make_unique>(domain1, 50000001); + auto value1InCommonDomain = value1->toCommonDomain(commonDomain); + + auto value1BackInRegularDomain = value1InCommonDomain->fromCommonDomain(domain1); + auto* value1BackInRegularDomainP = dynamic_cast*>(value1BackInRegularDomain.get()); + ASSERT_EQ(value1BackInRegularDomainP->getValue(), value1->getValue()); +} + +TEST_F(DomainValueTest, NonRepresentibleConversions) +{ + std::string commonEpochString = "1970-01-01T00:00:00+00:00"; + std::chrono::system_clock::time_point commonEpoch = daq::reader::parseEpoch(commonEpochString); + daq::RatioPtr commonResolution = daq::Ratio(1, 1000000); + daq::RatioPtr coarseResolution = daq::Ratio(1, 1000); + + daq::DomainInfo commonDomain = {commonEpoch, commonResolution}; + daq::DomainInfo coarseDomain = {commonEpoch, coarseResolution}; + + { + std::unique_ptr valueInFine = std::make_unique>(commonDomain, 1002); // 1002 / 1 000 000 + auto valueInCoarse = valueInFine->fromCommonDomain(coarseDomain); // 1 / 1 000 + auto valueBackInFine = valueInCoarse->toCommonDomain(commonDomain); // 1 000 / 1 000 000 + + // During a lossful conversion fine1->coarse->fine2 where coarse == fine2 it may be fine2 != fine1. Currently, the nearest tick is taken in the coarse domain. + // It may be required that if we find index that satisfies domain[index] >= coarse we found the correct sample. + // Rounding down case + ASSERT_FALSE(*valueInFine < *valueBackInFine); + } + + { + std::unique_ptr valueInFine = std::make_unique>(commonDomain, 998); // 998 / 1 000 000 + auto valueInCoarse = valueInFine->fromCommonDomain(coarseDomain); // 1 / 1 000 + auto valueBackInFine = valueInCoarse->toCommonDomain(commonDomain); // 1 000 / 1 000 000 + + // Rounding up case + ASSERT_TRUE(*valueInFine < *valueBackInFine); + } +} + +TEST_F(DomainValueTest, NonRepresentibleConversions2) +{ + std::string commonEpochString = "1970-01-01T00:00:00+00:00"; + std::chrono::system_clock::time_point commonEpoch = daq::reader::parseEpoch(commonEpochString); + daq::RatioPtr commonResolution = daq::Ratio(1, 7); + daq::RatioPtr coarseResolution = daq::Ratio(1, 3); + + daq::DomainInfo commonDomain = {commonEpoch, commonResolution}; + daq::DomainInfo coarseDomain = {commonEpoch, coarseResolution}; + + { + std::unique_ptr valueInFine = std::make_unique>(commonDomain, 2); + auto valueInCoarse = valueInFine->fromCommonDomain(coarseDomain); + auto valueBackInFine = valueInCoarse->toCommonDomain(commonDomain); + + ASSERT_FALSE(*valueInFine < *valueBackInFine || *valueBackInFine < *valueInFine); + } + + { + std::unique_ptr valueInCoarse = std::make_unique>(coarseDomain, 5); + auto valueInFine = valueInCoarse->toCommonDomain(commonDomain); + auto valueBackInCoarse = valueInFine->fromCommonDomain(coarseDomain); + + ASSERT_FALSE(*valueInCoarse < *valueBackInCoarse || *valueBackInCoarse < *valueInCoarse); + } +} \ No newline at end of file diff --git a/core/opendaq/reader/tests/test_multi_reader.cpp b/core/opendaq/reader/tests/test_multi_reader.cpp index 20974817df..2115665a83 100644 --- a/core/opendaq/reader/tests/test_multi_reader.cpp +++ b/core/opendaq/reader/tests/test_multi_reader.cpp @@ -955,6 +955,70 @@ TEST_F(MultiReaderTest, Clock10kHzDelta10) ASSERT_THAT(time[2], ElementsAreArray(time[0])); } +TEST_F(MultiReaderTest, Clock15MHzFromEpoch) +{ + constexpr const auto NUM_SIGNALS = 3; + + // prevent vector from re-allocating, so we have "stable" pointers + readSignals.reserve(3); + + Int clock = 15000000; + Int startOffest = 1781269748ll * clock; + auto& sig0 = addSignal(startOffest, 6093750, createDomainSignal("1970-01-01T00:00:00+00:00", Ratio(1, clock))); + auto& sig1 = addSignal(startOffest, 2812500, createDomainSignal("1970-01-01T00:00:00+00:00", Ratio(1, clock))); + auto& sig2 = addSignal(startOffest, 3750000, createDomainSignal("1970-01-01T00:00:00+00:00", Ratio(1, clock))); + + auto multi = MultiReaderBuilder() + .setStartOnFullUnitOfDomain(true) + .setInputPortNotificationMethod(PacketReadyNotification::SameThread) + .addSignals(signalsToList()) + .build(); + + { + SizeT count{0}; + auto status = multi.read(nullptr, &count); + ASSERT_EQ(status.getReadStatus(), ReadStatus::Event); + } + + auto available = multi.getAvailableCount(); + ASSERT_EQ(available, 0u); + + sig0.createAndSendPacket(0); + sig1.createAndSendPacket(0); + sig2.createAndSendPacket(0); + + sig0.createAndSendPacket(1); + sig1.createAndSendPacket(1); + sig2.createAndSendPacket(1); + + sig0.createAndSendPacket(2); + sig1.createAndSendPacket(2); + sig2.createAndSendPacket(2); + + available = multi.getAvailableCount(); + ASSERT_EQ(available, 2812500 * 3); + + constexpr const SizeT SAMPLES = 5u; + + std::array values{}; + std::array domain{}; + + void* valuesPerSignal[NUM_SIGNALS]{values[0], values[1], values[2]}; + void* domainPerSignal[NUM_SIGNALS]{domain[0], domain[1], domain[2]}; + + SizeT count{SAMPLES}; + multi.readWithDomain(valuesPerSignal, domainPerSignal, &count); + + ASSERT_EQ(count, SAMPLES); + + std::array time{}; + printData(SAMPLES, time, values, domain); + + ASSERT_THAT(time[1], ElementsAreArray(time[0])); + ASSERT_THAT(time[2], ElementsAreArray(time[0])); +} + + TEST_F(MultiReaderTest, Clock10kHzDelta10Relative) { constexpr const auto NUM_SIGNALS = 3; @@ -5234,3 +5298,76 @@ TEST_F(MultiReaderTest, UsedUnusedInput) ASSERT_TRUE(status.getValid()); } } + +TEST_F(MultiReaderTest, CheckSpecificCase) +{ + readSignals.reserve(2); + + auto sig0 = addSignal(0, 2, createDomainSignal("2022-09-27T00:02:03+00:00", Ratio(1, 1000), LinearDataRule(1, 0), nullptr)); // 1000 Hz + auto sig1 = addSignal(0, 1, createDomainSignal("2022-09-27T00:02:03+00:00", Ratio(1, 1000), LinearDataRule(5, 0), nullptr)); // 200 Hz + + const MultiReaderPtr multiReader = MultiReaderBuilder() + .setInputPortNotificationMethod(PacketReadyNotification::SameThread) + .addSignals(signalsToList()) + .setValueReadType(SampleType::Float64) + .setDomainReadType(SampleType::Int64) + .build(); + + { + SizeT count{0}; + auto status = multiReader.read(nullptr, &count); + ASSERT_EQ(status.getReadStatus(), ReadStatus::Event); + } + + constexpr size_t numberOfSamplesToRead = 12; + double dataFirstSignal[2*numberOfSamplesToRead]; + double dataSecondSignal[2*numberOfSamplesToRead]; + double* data[2]{dataFirstSignal, dataSecondSignal}; + + sig0.createAndSendPacket(0, true); + sig0.createAndSendPacket(1, true); + sig0.createAndSendPacket(2, true); + sig0.createAndSendPacket(3, true); + sig0.createAndSendPacket(4, true); + sig0.createAndSendPacket(5, true); + sig0.setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).setUnit(Unit("A", -1, "ampere", "current")).build()); + sig0.createAndSendPacket(6, true); + sig0.createAndSendPacket(7, true); + sig0.createAndSendPacket(8, true); + + sig1.createAndSendPacket(0, true); + sig1.createAndSendPacket(1, true); + sig1.createAndSendPacket(2, true); + sig1.createAndSendPacket(3, true); + sig1.createAndSendPacket(4, true); + sig1.createAndSendPacket(5, true); + sig1.createAndSendPacket(6, true); + sig1.createAndSendPacket(7, true); + + { + auto available = multiReader.getAvailableCount(); + ASSERT_EQ(available, 10u); + + SizeT count = numberOfSamplesToRead; + auto status = multiReader.read(data, &count); + ASSERT_EQ(count, 10u); + } + + { + auto available = multiReader.getAvailableCount(); + ASSERT_EQ(available, 0); + + SizeT count{2}; + auto status = multiReader.read(data, &count); + ASSERT_EQ(status.getReadStatus(), ReadStatus::Ok); + } + + { + auto available = multiReader.getAvailableCount(); + ASSERT_EQ(available, 0); + + SizeT count{0}; + auto status = multiReader.read(data, &count); + ASSERT_EQ(status.getReadStatus(), ReadStatus::Ok); + } +} diff --git a/core/opendaq/reader/tests/test_queue_reader.cpp b/core/opendaq/reader/tests/test_queue_reader.cpp new file mode 100644 index 0000000000..c87bba1e59 --- /dev/null +++ b/core/opendaq/reader/tests/test_queue_reader.cpp @@ -0,0 +1,1054 @@ +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "reader_common.h" + +#include +#include +#include + +class QueueReaderTest : public ReaderTest<> +{ +public: + using Super = ReaderTest<>; + +protected: + void SetUp() override + { + Super::SetUp(); + + domainSignal = Signal(context, nullptr, "timeSig"); + signal.setDomainSignal(domainSignal); + } + + void setDomainDescriptor(const DataDescriptorPtr& descriptor) + { + domainSignal.setDescriptor(descriptor); + } + + void setValueDescriptor(const DataDescriptorPtr& descriptor) + { + signal.setDescriptor(descriptor); + } + + void setPacketSize(SizeT size) + { + packetSize = size; + } + + void setOffsetDelta(Int newOffset, Int newDelta) + { + offset = newOffset; + delta = newDelta; + samples = 0; + } + + Int getOffset() const + { + return offset; + } + + void sendNextPacket() + { + const auto domainPacket = DataPacket(domainSignal.getDescriptor(), packetSize, offset); + const auto valuePacket = DataPacketWithDomain(domainPacket, signal.getDescriptor(), packetSize); + auto* d = static_cast(valuePacket.getRawData()); + for (size_t i = 0; i < packetSize; ++i) + { + d[i] = static_cast(samples + i); + } + + domainSignal.sendPacket(domainPacket); + signal.sendPacket(valuePacket); + + samples += packetSize; + offset += packetSize * delta; + } + +protected: + Int offset = 0; + Int delta = 1; + SizeT packetSize = 1; + + SizeT samples = 0; + + SignalConfigPtr domainSignal; +}; + +void assertReaderAtDomainValue(QueueReader& reader, Int tick) +{ + auto start = reader.getFirstSampleDomainValue(); + auto* startP = dynamic_cast*>(start.get()); + ASSERT_EQ(startP->getValue(), tick); +} + +TEST_F(QueueReaderTest, AdvancePastEnd) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 10000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("2022-09-27T00:02:03+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + for (int c = 0; c < 3; ++c) + { + sendNextPacket(); + } + + ASSERT_TRUE(reader.hasPendingEvents()); + + ASSERT_EQ(reader.getAvailableSamples(), 3u * packetSize); + + auto start = reader.getFirstSampleDomainValue(); + auto* startP = dynamic_cast*>(start.get()); + + ASSERT_TRUE(startP != nullptr); + + ASSERT_EQ(startP->getValue(), 500u); + + std::unique_ptr domainValue = std::make_unique>(reader.getDomainInfo(), 512); + AdvanceResult result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::Success); + + auto start2 = reader.getFirstSampleDomainValue(); + ASSERT_EQ(*domainValue, *start2); + + ASSERT_TRUE(reader.hasPendingEvents()); + auto event = reader.popFrontEvent(); + ASSERT_EQ(event.getType(), PacketType::Event); + auto params = event.getParameters(); + const DataDescriptorPtr domainFromEvent = params[event_packet_param::DOMAIN_DATA_DESCRIPTOR]; + ASSERT_EQ(domainFromEvent.getTickResolution().getDenominator(), sampleRate); + ASSERT_FALSE(reader.hasPendingEvents()); + + domainValue = std::make_unique>(reader.getDomainInfo(), 100512); + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::NeedMoreData); + + start2 = reader.getFirstSampleDomainValue(); + ASSERT_EQ(start2.get(), nullptr); +} + +TEST_F(QueueReaderTest, DomainChangeDetection) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 10000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + sendNextPacket(); + + ASSERT_TRUE(reader.hasPendingEvents()); + + ASSERT_EQ(reader.getAvailableSamples(), packetSize); + + auto start = reader.getFirstSampleDomainValue(); + auto* startP = dynamic_cast*>(start.get()); + + ASSERT_TRUE(startP != nullptr); + + ASSERT_EQ(startP->getValue(), 500u); + + std::unique_ptr domainValue = std::make_unique>(reader.getDomainInfo(), 512); + AdvanceResult result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::NeedMoreData); + + auto start2 = reader.getFirstSampleDomainValue(); + ASSERT_TRUE(start2 == nullptr); + + sendNextPacket(); + sendNextPacket(); + + domainValue = std::make_unique>(reader.getDomainInfo(), 512); + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::Success); + + // The descriptor change event when the sig was connected to port + ASSERT_TRUE(reader.hasPendingEvents()); + auto eventPacket = reader.popFrontEvent(); + ASSERT_FALSE(reader.hasPendingEvents()); + ASSERT_EQ(reader.getSampleRate(), sampleRate); + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(10, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 10); + + ASSERT_EQ(reader.getAvailableSamples(), 3u); + + domainValue = std::make_unique>(reader.getDomainInfo(), 525); + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::DomainChanged); + + // Event from changing the descriptor mid operation + ASSERT_TRUE(reader.hasPendingEvents()); + + eventPacket = reader.popFrontEvent(); + ASSERT_FALSE(reader.hasPendingEvents()); + ASSERT_EQ(reader.getSampleRate(), sampleRate / 10); + ASSERT_TRUE(reader.isValid()); +} + +TEST_F(QueueReaderTest, OriginParsing) +{ + std::string origin = "1970-01-01T00:01:00+0000"; + auto epoch = reader::tryParseEpoch(origin); + ASSERT_TRUE(epoch.has_value()); + + origin = "1970-01-01T00:01:00+00:00"; + epoch = reader::tryParseEpoch(origin); + ASSERT_TRUE(epoch.has_value()); + + origin = "abc"; + epoch = reader::tryParseEpoch(origin); + ASSERT_FALSE(epoch.has_value()); + + origin = "1970-01-01T00:01:00+00:00abc"; + epoch = reader::tryParseEpoch(origin); + ASSERT_FALSE(epoch.has_value()); +} + +TEST_F(QueueReaderTest, CreateBeforeConnection) +{ + auto inputPort = InputPort(context, nullptr, "port", true); + + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + std::unique_ptr domainValue = + std::make_unique>(DomainInfo{std::chrono::system_clock::time_point{}, Ratio(1, 1000)}, 512); + + bool valid; + ASSERT_NO_THROW(valid = reader.isValid()); + ASSERT_FALSE(valid); + + ASSERT_THROW(reader.getDomainInfo(), InvalidOperationException); + ASSERT_THROW(reader.getFirstSampleDomainValue(), InvalidOperationException); + ASSERT_THROW(reader.advanceToDomainValue(domainValue.get()), InvalidOperationException); + ASSERT_THROW(reader.getSampleRate(), InvalidOperationException); + ASSERT_THROW(reader.dropOutdatedPacketSegments(), InvalidOperationException); + ASSERT_THROW(reader.hasPendingEvents(), InvalidOperationException); + ASSERT_THROW(reader.popFrontEvent(), InvalidOperationException); +} + +TEST_F(QueueReaderTest, CreateBeforeConnectionRecovery) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 10000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + ASSERT_FALSE(reader.isValid()); + ASSERT_THROW(reader.getDomainInfo(), InvalidOperationException); + + inputPort.connect(signal); + reader.updateConnection(); + + std::unique_ptr domainValue = + std::make_unique>(DomainInfo{std::chrono::system_clock::time_point{}, Ratio(1, 1000)}, 512); + + bool valid; + ASSERT_NO_THROW(valid = reader.isValid()); + ASSERT_TRUE(valid); + + ASSERT_NO_THROW(reader.getDomainInfo()); + ASSERT_NO_THROW(reader.getFirstSampleDomainValue()); + ASSERT_NO_THROW(reader.advanceToDomainValue(domainValue.get())); + Int sr; + ASSERT_NO_THROW(sr = reader.getSampleRate()); + ASSERT_EQ(sr, sampleRate); + ASSERT_NO_THROW(reader.dropOutdatedPacketSegments()); + ASSERT_NO_THROW(reader.hasPendingEvents()); + ASSERT_NO_THROW(reader.popFrontEvent()); +} + +TEST_F(QueueReaderTest, InvalidDomainAndBack) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 10000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("abc") // Invalid origin + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.setNotificationMethod(PacketReadyNotification::SameThread); + + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + ASSERT_FALSE(reader.isValid()); + ASSERT_THROW(reader.getDomainInfo(), InvalidOperationException); + + inputPort.connect(signal); + reader.updateConnection(); + + bool valid; + ASSERT_NO_THROW(valid = reader.isValid()); + ASSERT_FALSE(valid); + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + ASSERT_TRUE(reader.isValid()); + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(3.5, 0)) // Non-integer delta + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + ASSERT_FALSE(reader.isValid()); + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + ASSERT_TRUE(reader.isValid()); + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(3, 0)) // Non-integer sample rate + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + ASSERT_FALSE(reader.isValid()); + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + ASSERT_TRUE(reader.isValid()); + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(ExplicitDataRule()) // Explicit data rule - shall be removed when resampling is added + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + ASSERT_FALSE(reader.isValid()); +} + +TEST_F(QueueReaderTest, MergeDomainAndValueChange) +{ + constexpr Int sampleRate = 10000; + auto domainDescriptor = DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build(); + + auto valueDescriptor = DataDescriptorBuilder().setSampleType(SampleType::Float64).setUnit(Unit("V", -1, "volt", "voltage")).build(); + + auto domainChangePacket = + DataDescriptorChangedEventPacket(descriptorToEventPacketParam(nullptr), descriptorToEventPacketParam(domainDescriptor)); + SignalEvent domainChange(domainChangePacket); + + ASSERT_EQ(domainChange.getType(), SignalEventType::DomainChanged); + + auto valueChangePacket = + DataDescriptorChangedEventPacket(descriptorToEventPacketParam(valueDescriptor), descriptorToEventPacketParam(nullptr)); + SignalEvent valueChange(valueChangePacket); + + ASSERT_EQ(valueChange.getType(), SignalEventType::ValueChanged); + + auto merged = domainChange.merge(valueChange); + ASSERT_TRUE(merged); + + ASSERT_EQ(domainChange.getType(), SignalEventType::DomainAndValueChanged); + + ASSERT_EQ(domainChange.getDomainDescriptor().getTickResolution().getDenominator(), sampleRate); + ASSERT_EQ(domainChange.getValueDescriptor().getUnit().getQuantity(), String("voltage")); +} + +TEST_F(QueueReaderTest, LatestDescriptorPreserved) +{ + constexpr Int sampleRate = 10000; + auto domainDescriptor = DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build(); + auto domainDescriptor2 = DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(10, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build(); + + auto domainChangePacket = + DataDescriptorChangedEventPacket(descriptorToEventPacketParam(nullptr), descriptorToEventPacketParam(domainDescriptor)); + SignalEvent domainChange(domainChangePacket); + ASSERT_EQ(domainChange.getType(), SignalEventType::DomainChanged); + NumberPtr delta = domainChange.getDomainDescriptor().getRule().getParameters()["delta"]; + ASSERT_EQ(delta.getIntValue(), 1u); + + auto domainChangePacket2 = + DataDescriptorChangedEventPacket(descriptorToEventPacketParam(nullptr), descriptorToEventPacketParam(domainDescriptor2)); + SignalEvent domainChange2(domainChangePacket2); + ASSERT_EQ(domainChange2.getType(), SignalEventType::DomainChanged); + delta = domainChange2.getDomainDescriptor().getRule().getParameters()["delta"]; + ASSERT_EQ(delta.getIntValue(), 10u); + + bool merged = domainChange.merge(domainChange2); + ASSERT_TRUE(merged); + + ASSERT_EQ(domainChange.getType(), SignalEventType::DomainChanged); + delta = domainChange.getDomainDescriptor().getRule().getParameters()["delta"]; + ASSERT_EQ(delta.getIntValue(), 10u); +} + +TEST_F(QueueReaderTest, GapEventsRefuseMerge) +{ + constexpr Int sampleRate = 10000; + auto domainDescriptor = DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build(); + + auto domainChangePacket = + DataDescriptorChangedEventPacket(descriptorToEventPacketParam(nullptr), descriptorToEventPacketParam(domainDescriptor)); + SignalEvent domainChange(domainChangePacket); + ASSERT_EQ(domainChange.getType(), SignalEventType::DomainChanged); + + auto gapPacket = ImplicitDomainGapDetectedEventPacket(0); + SignalEvent gapChange(gapPacket); + ASSERT_EQ(gapChange.getType(), SignalEventType::Gap); + + auto merged = domainChange.merge(gapChange); + ASSERT_FALSE(merged); +} + +TEST_F(QueueReaderTest, CheckAdvanceDomainEdgeCases) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 10000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + sendNextPacket(); // [500 - 504] + sendNextPacket(); + sendNextPacket(); // [510 - 514] + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(10, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 10); + sendNextPacket(); // [515 - 555] + sendNextPacket(); // [565 - 605] + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(2, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 2); + + sendNextPacket(); // [615 - 623] + sendNextPacket(); // [625 - 633] + // Establish a queue, now test queue handling + + // Initial data segment + ASSERT_TRUE(reader.hasPendingEvents()); + ASSERT_EQ(reader.getAvailableSamples(), 3 * packetSize); // First three packets worth of samples + + auto event = reader.popFrontEvent(); + ASSERT_FALSE(reader.hasPendingEvents()); + + DataDescriptorPtr descriptor = event.getParameters()[event_packet_param::DOMAIN_DATA_DESCRIPTOR]; + NumberPtr delta = descriptor.getRule().getParameters()["delta"]; + ASSERT_EQ(delta.getIntValue(), 1u); + + auto start = reader.getFirstSampleDomainValue(); + auto* startP = dynamic_cast*>(start.get()); + ASSERT_EQ(startP->getValue(), 500u); + // End Initial data segment + + // Second data segment + std::unique_ptr domainValue = std::make_unique>(reader.getDomainInfo(), 515); + AdvanceResult result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::DomainChanged); + assertReaderAtDomainValue(reader, 515); // First available sample is the first sample of the "unopened" data packet + + ASSERT_TRUE(reader.hasPendingEvents()); + + event = reader.popFrontEvent(); + ASSERT_FALSE(reader.hasPendingEvents()); + + descriptor = event.getParameters()[event_packet_param::DOMAIN_DATA_DESCRIPTOR]; + delta = descriptor.getRule().getParameters()["delta"]; + ASSERT_EQ(delta.getIntValue(), 10u); // Check that we got the second descriptor + + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::Success); // Advancing for the second time will result in a success + assertReaderAtDomainValue(reader, 515); + + ASSERT_EQ(reader.getSampleRate(), sampleRate/10); + + domainValue = std::make_unique>(reader.getDomainInfo(), 605); + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::Success); // Can advance to the last sample in the data segment + assertReaderAtDomainValue(reader, 605); + // End Second data segment + + // Third data segment + domainValue = std::make_unique>(reader.getDomainInfo(), 615); + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::DomainChanged); // The first sample in the next sample only through domain change + assertReaderAtDomainValue(reader, 615); + + ASSERT_TRUE(reader.hasPendingEvents()); + + event = reader.popFrontEvent(); + ASSERT_FALSE(reader.hasPendingEvents()); + + descriptor = event.getParameters()[event_packet_param::DOMAIN_DATA_DESCRIPTOR]; + delta = descriptor.getRule().getParameters()["delta"]; + ASSERT_EQ(delta.getIntValue(), 2u); // Check that we got the third descriptor + + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::Success); // Advancing for the second time will result in a success + assertReaderAtDomainValue(reader, 615); + + ASSERT_EQ(reader.getSampleRate(), sampleRate/2); + + domainValue = std::make_unique>(reader.getDomainInfo(), 633); + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::Success); // Can advance to the last sample in the data segment + assertReaderAtDomainValue(reader, 633); + + domainValue = std::make_unique>(reader.getDomainInfo(), 650); + result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::NeedMoreData); + + // Empty queue + ASSERT_EQ(reader.getAvailableSamples(), 0u); + auto first = reader.getFirstSampleDomainValue(); + ASSERT_EQ(first.get(), nullptr); + ASSERT_FALSE(reader.hasPendingEvents()); + event = reader.popFrontEvent(); + ASSERT_EQ(event.getObject(), nullptr); +} + +TEST_F(QueueReaderTest, DropOutdatedPacketSegments) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 10000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + sendNextPacket(); // [500 - 504] + sendNextPacket(); + sendNextPacket(); // [510 - 514] + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(10, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 10); + sendNextPacket(); // [515 - 555] + sendNextPacket(); // [565 - 605] + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(2, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 2); + + sendNextPacket(); // [615 - 623] + sendNextPacket(); // [625 - 633] + // Establish a queue, now test queue handling + + reader.dropOutdatedPacketSegments(); + + ASSERT_TRUE(reader.hasPendingEvents()); + + auto event = reader.popFrontEvent(); + ASSERT_FALSE(reader.hasPendingEvents()); + + DataDescriptorPtr descriptor = event.getParameters()[event_packet_param::DOMAIN_DATA_DESCRIPTOR]; + NumberPtr delta = descriptor.getRule().getParameters()["delta"]; + ASSERT_EQ(delta.getIntValue(), 2u); + + assertReaderAtDomainValue(reader, 615); +} + +TEST_F(QueueReaderTest, DropLeftoverSegment) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 1000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + sendNextPacket(); // [500 - 504] + sendNextPacket(); + sendNextPacket(); // [510 - 514] + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(10, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 10); + sendNextPacket(); // [515 - 555] + sendNextPacket(); // [565 - 605] + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(2, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 2); + + sendNextPacket(); // [615 - 623] + sendNextPacket(); // [625 - 633] + // Establish a queue, now test queue handling + ASSERT_TRUE(reader.hasPendingEvents()); + auto event = reader.popFrontEvent(); + ASSERT_TRUE(event.assigned()); + ASSERT_FALSE(reader.hasPendingEvents()); + + ASSERT_EQ(reader.getAvailableSamples(), 15); + ASSERT_THROW(reader.setSampleRateDivider(0), InvalidParameterException); + + ASSERT_NO_THROW(reader.setSampleRateDivider(10)); + ASSERT_EQ(reader.getAvailableSamples(), 150u); + ASSERT_THROW(reader.dropLeftoverSegment(13), InvalidStateException); + + ASSERT_FALSE(reader.dropLeftoverSegment(10)); + ASSERT_FALSE(reader.dropLeftoverSegment(150)); + ASSERT_TRUE(reader.dropLeftoverSegment(200)); + + ASSERT_TRUE(reader.hasPendingEvents()); + ASSERT_THROW(reader.dropLeftoverSegment(10), InvalidStateException); // Cannot drop with pending events + event = reader.popFrontEvent(); + ASSERT_TRUE(event.assigned()); + ASSERT_EQ(event.getEventId(), event_packet_id::IMPLICIT_DOMAIN_GAP_DETECTED); + ASSERT_TRUE(reader.hasPendingEvents()); + event = reader.popFrontEvent(); + ASSERT_EQ(event.getEventId(), event_packet_id::DATA_DESCRIPTOR_CHANGED); + ASSERT_FALSE(reader.hasPendingEvents()); + + ASSERT_EQ(reader.getAvailableSamples(), 100u); + ASSERT_NO_THROW(reader.setSampleRateDivider(100)); + ASSERT_EQ(reader.getAvailableSamples(), 1000u); + + ASSERT_FALSE(reader.dropLeftoverSegment(100)); + ASSERT_FALSE(reader.dropLeftoverSegment(1000)); + ASSERT_THROW(reader.dropLeftoverSegment(1001), InvalidStateException); + ASSERT_TRUE(reader.dropLeftoverSegment(1100)); + ASSERT_THROW(reader.dropLeftoverSegment(100), InvalidStateException); // pending events + + ASSERT_TRUE(reader.hasPendingEvents()); + event = reader.popFrontEvent(); + ASSERT_TRUE(event.assigned()); + ASSERT_EQ(event.getEventId(), event_packet_id::IMPLICIT_DOMAIN_GAP_DETECTED); + ASSERT_TRUE(reader.hasPendingEvents()); + event = reader.popFrontEvent(); + ASSERT_EQ(event.getEventId(), event_packet_id::DATA_DESCRIPTOR_CHANGED); + ASSERT_FALSE(reader.hasPendingEvents()); + + reader.setSampleRateDivider(2); + ASSERT_EQ(reader.getAvailableSamples(), 20u); + ASSERT_FALSE(reader.dropLeftoverSegment(1000)); // Don't drop if unfinished segment +} + +TEST_F(QueueReaderTest, LeftoverSegmentPartialPackets) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 10000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 200; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + sendNextPacket(); // [500 - 699] + sendNextPacket(); // [700 - 899] + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(2, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 2); + setPacketSize(packetSize / 2); + + sendNextPacket(); + + ASSERT_TRUE(reader.hasPendingEvents()); + reader.popFrontEvent(); + + std::unique_ptr domainValue = std::make_unique>(reader.getDomainInfo(), 870); + AdvanceResult result = reader.advanceToDomainValue(domainValue.get()); + ASSERT_EQ(result, AdvanceResult::Success); + + ASSERT_FALSE(reader.dropLeftoverSegment(10)); + ASSERT_FALSE(reader.dropLeftoverSegment(30)); + ASSERT_TRUE(reader.dropLeftoverSegment(31)); +} + +TEST_F(QueueReaderTest, TestReading) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 1000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + sendNextPacket(); // [500 - 504] + sendNextPacket(); + sendNextPacket(); // [510 - 514] + sendNextPacket(); // [515 - 519] + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(10, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 10); + sendNextPacket(); // [515 - 555] + sendNextPacket(); // [565 - 605] + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(2, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 2); + + sendNextPacket(); // [615 - 623] + sendNextPacket(); // [625 - 633] + // Establish a queue, now test queue handling + + SizeT count = 10; + ASSERT_EQ(reader.read(nullptr, nullptr, &count), AdvanceResult::Error); + ASSERT_TRUE(reader.hasPendingEvents()); + reader.popFrontEvent(); + ASSERT_EQ(reader.getAvailableSamples(), 20u); + + std::array buffer; + count = 20; + auto result = reader.read(buffer.data(), nullptr, &count); + ASSERT_EQ(result, AdvanceResult::Success); + ASSERT_EQ(count, 20u); + for (size_t i = 0; i < 20; ++i) + { + ASSERT_EQ(buffer[i], static_cast(i)); + } + + ASSERT_TRUE(reader.hasPendingEvents()); + auto event = reader.popFrontEvent(); + ASSERT_EQ(event.getEventId(), event_packet_id::DATA_DESCRIPTOR_CHANGED); + + reader.setSampleRateDivider(10); + ASSERT_EQ(reader.getAvailableSamples(), 100); + + count = 80; + result = reader.read(buffer.data(), nullptr, &count); + ASSERT_EQ(result, AdvanceResult::Success); + ASSERT_EQ(count, 80u); + ASSERT_FALSE(reader.hasPendingEvents()); + + ASSERT_TRUE(reader.dropLeftoverSegment(40)); + ASSERT_TRUE(reader.hasPendingEvents()); + event = reader.popFrontEvent(); + ASSERT_EQ(event.getEventId(), event_packet_id::IMPLICIT_DOMAIN_GAP_DETECTED); + event = reader.popFrontEvent(); + ASSERT_EQ(event.getEventId(), event_packet_id::DATA_DESCRIPTOR_CHANGED); + + reader.setSampleRateDivider(2); + ASSERT_EQ(reader.getAvailableSamples(), 20u); + ASSERT_FALSE(reader.dropLeftoverSegment(8)); + ASSERT_FALSE(reader.dropLeftoverSegment(5000)); // Non-delimited segment +} + +TEST_F(QueueReaderTest, ReadingEdgeCases) +{ + // Domain (time) signal: Int64, linear rule. + constexpr Int sampleRate = 1000; + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(1, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setValueDescriptor(DataDescriptorBuilder().setSampleType(SampleType::Float64).build()); + + setOffsetDelta(500, 1); + + const size_t packetSize = 5; + setPacketSize(packetSize); + + auto inputPort = InputPort(context, nullptr, "port", true); + inputPort.connect(signal); + QueueReader reader = QueueReader(inputPort, SampleType::Float64, SampleType::Int64, ReadMode::Scaled, loggerComponent, false); + + sendNextPacket(); // [500 - 504] + sendNextPacket(); + sendNextPacket(); // [510 - 514] + sendNextPacket(); // [515 - 519] + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(10, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 10); + sendNextPacket(); // [515 - 555] + sendNextPacket(); // [565 - 605] + + setDomainDescriptor(DataDescriptorBuilder() + .setSampleType(SampleType::Int64) + .setTickResolution(Ratio(1, sampleRate)) + .setOrigin("1970-01-01T00:00:00+00:00") + .setRule(LinearDataRule(2, 0)) + .setUnit(Unit("s", -1, "second", "time")) + .build()); + + setOffsetDelta(getOffset(), 2); + + sendNextPacket(); // [615 - 623] + sendNextPacket(); // [625 - 633] + // Establish a queue, now test queue handling + + std::array buffer; + + reader.popFrontEvent(); + ASSERT_EQ(reader.getAvailableSamples(), 20u); + auto result = reader.read(nullptr, nullptr, nullptr); + ASSERT_EQ(result, AdvanceResult::Error); + SizeT count = 18; + result = reader.read(nullptr, nullptr, &count); + ASSERT_EQ(result, AdvanceResult::Success); + ASSERT_EQ(count, 18u); + + ASSERT_EQ(reader.getAvailableSamples(), 2u); + reader.setSampleRateDivider(10); + count = 15; + result = reader.read(buffer.data(), nullptr, &count); + ASSERT_EQ(result, AdvanceResult::Error); + ASSERT_EQ(count, 0u); + count = 20; + result = reader.read(buffer.data(), nullptr, &count); + ASSERT_EQ(result, AdvanceResult::Success); + ASSERT_EQ(count, 20u); + + reader.popFrontEvent(); + count = 0; + ASSERT_EQ(reader.read(buffer.data(), nullptr, &count), AdvanceResult::Success); + ASSERT_EQ(count, 0u); + + reader.dropOutdatedPacketSegments(); + reader.popFrontEvent(); + ASSERT_EQ(reader.getAvailableSamples(), 100u); + count = 200; + ASSERT_EQ(reader.read(nullptr, nullptr, &count), AdvanceResult::NeedMoreData); +} \ No newline at end of file diff --git a/core/opendaq/reader/tests/test_typed_reading.cpp b/core/opendaq/reader/tests/test_typed_reading.cpp new file mode 100644 index 0000000000..b2c9b90f43 --- /dev/null +++ b/core/opendaq/reader/tests/test_typed_reading.cpp @@ -0,0 +1,159 @@ +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include + +using TypedReadingTest = testing::Test; + +TEST_F(TypedReadingTest, LinearRuleDomainReading) +{ + daq::RatioPtr resolution = daq::Ratio(1, 1'000'000'000); + daq::DataDescriptorPtr domainDescriptor = daq::DataDescriptorBuilder() + .setSampleType(daq::SampleType::UInt64) + .setRule(daq::LinearDataRule(1'000'000, 0)) + .setTickResolution(resolution) + .setUnit(daq::Unit("s", -1, "seconds", "time")) + .setOrigin("1970-01-01T00:00:00") + .setName("Time") + .build(); + + daq::DataPacketPtr domainPacket = daq::DataPacket(domainDescriptor, 100, 115); + + daq::ReadLayout readLayout = daq::TypedReadingUtils::createReadLayout(domainDescriptor); + daq::DomainInfo domainInfo = daq::DomainInfo::fromDescriptor(domainDescriptor); + std::unique_ptr domainStart = + daq::TypedReadingUtils::readDomainValue(daq::SampleType::UInt64, daq::SampleType::UInt64, readLayout, domainPacket, 12, domainInfo); + + auto* domainStartP = dynamic_cast*>(domainStart.get()); + ASSERT_FALSE(domainStartP == nullptr); + + ASSERT_EQ(domainStartP->getValue(), 12'000'115u); +} + +TEST_F(TypedReadingTest, ExplicitRuleDomainReading) +{ + daq::RatioPtr resolution = daq::Ratio(1, 1'000'000'000); + daq::DataDescriptorPtr domainDescriptor = daq::DataDescriptorBuilder() + .setSampleType(daq::SampleType::UInt64) + .setTickResolution(resolution) + .setUnit(daq::Unit("s", -1, "seconds", "time")) + .setOrigin("1970-01-01T00:00:00") + .setName("Time") + .build(); + + constexpr daq::SizeT packetSize = 100; + daq::DataPacketPtr domainPacket = daq::DataPacket(domainDescriptor, packetSize, 0); + daq::UInt* data = static_cast(domainPacket.getRawData()); + for (daq::SizeT i = 0; i < packetSize; ++i) + { + data[i] = 112 + 1'000'000 * i; + } + + daq::ReadLayout readLayout = daq::TypedReadingUtils::createReadLayout(domainDescriptor); + daq::DomainInfo domainInfo = daq::DomainInfo::fromDescriptor(domainDescriptor); + std::unique_ptr domainStart = + daq::TypedReadingUtils::readDomainValue(daq::SampleType::UInt64, daq::SampleType::UInt64, readLayout, domainPacket, 12, domainInfo); + + auto* domainStartP = dynamic_cast*>(domainStart.get()); + ASSERT_FALSE(domainStartP == nullptr); + + ASSERT_EQ(domainStartP->getValue(), 12'000'112u); +} + +TEST_F(TypedReadingTest, LinearRuleFindDomain) +{ + daq::RatioPtr resolution = daq::Ratio(1, 1'000'000'000); + daq::DataDescriptorPtr domainDescriptor = daq::DataDescriptorBuilder() + .setSampleType(daq::SampleType::UInt64) + .setRule(daq::LinearDataRule(1'000'000, 0)) + .setTickResolution(resolution) + .setUnit(daq::Unit("s", -1, "seconds", "time")) + .setOrigin("1970-01-01T00:00:00") + .setName("Time") + .build(); + + daq::DataPacketPtr domainPacket = daq::DataPacket(domainDescriptor, 100, 1'000'000'115); + + daq::ReadLayout readLayout = daq::TypedReadingUtils::createReadLayout(domainDescriptor); + daq::DomainInfo domainInfo = daq::DomainInfo::fromDescriptor(domainDescriptor); + std::unique_ptr domainTarget = std::make_unique>(domainInfo, 1'009'000'115); + daq::SizeT index = daq::TypedReadingUtils::findDomainValue( + daq::SampleType::UInt64, daq::SampleType::UInt64, readLayout, domainPacket, domainTarget.get()); + + ASSERT_EQ(index, 9u); +} + +TEST_F(TypedReadingTest, ExplicitRuleFindDomain) +{ + daq::RatioPtr resolution = daq::Ratio(1, 1'000'000'000); + daq::DataDescriptorPtr domainDescriptor = daq::DataDescriptorBuilder() + .setSampleType(daq::SampleType::UInt64) + .setTickResolution(resolution) + .setUnit(daq::Unit("s", -1, "seconds", "time")) + .setOrigin("1970-01-01T00:00:00") + .setName("Time") + .build(); + + constexpr daq::SizeT packetSize = 100; + daq::DataPacketPtr domainPacket = daq::DataPacket(domainDescriptor, packetSize, 0); + daq::UInt* data = static_cast(domainPacket.getRawData()); + for (daq::SizeT i = 0; i < packetSize; ++i) + { + data[i] = 1'000'000'112 + 1'000'000 * i; + } + + daq::ReadLayout readLayout = daq::TypedReadingUtils::createReadLayout(domainDescriptor); + daq::DomainInfo domainInfo = daq::DomainInfo::fromDescriptor(domainDescriptor); + std::unique_ptr domainTarget = std::make_unique>(domainInfo, 1'012'000'112); + daq::SizeT index = daq::TypedReadingUtils::findDomainValue( + daq::SampleType::UInt64, daq::SampleType::UInt64, readLayout, domainPacket, domainTarget.get()); + + ASSERT_EQ(index, 12u); +} + +TEST_F(TypedReadingTest, ExplicitRuleReadData) +{ + daq::RatioPtr resolution = daq::Ratio(1, 1'000'000'000); + daq::DataDescriptorPtr domainDescriptor = daq::DataDescriptorBuilder() + .setSampleType(daq::SampleType::UInt64) + .setTickResolution(resolution) + .setUnit(daq::Unit("s", -1, "seconds", "time")) + .setOrigin("1970-01-01T00:00:00") + .setName("Time") + .build(); + + constexpr daq::SizeT packetSize = 15; + daq::DataPacketPtr domainPacket = daq::DataPacket(domainDescriptor, packetSize, 0); + daq::UInt* data = static_cast(domainPacket.getRawData()); + for (daq::SizeT i = 0; i < packetSize; ++i) + { + data[i] = 1'000'000'119 + 1'000'000 * i; + } + + daq::ReadLayout readLayout = daq::TypedReadingUtils::createReadLayout(domainDescriptor); + daq::DomainInfo domainInfo = daq::DomainInfo::fromDescriptor(domainDescriptor); + + std::vector buffer(packetSize); + for (int i = 0; i < packetSize; ++i) + { + void* bufferP = buffer.data(); + daq::SizeT count = packetSize - i; + daq::ErrCode err = daq::TypedReadingUtils::readData( + daq::SampleType::UInt64, daq::SampleType::UInt64, true, readLayout, domainPacket.getRawData(), i, &bufferP, count); + + for (daq::SizeT j = 0; j < count; ++j) + { + ASSERT_EQ(buffer[j], data[i + j]); + } + } +} \ No newline at end of file