DataStreamInstrument.h

Go to the documentation of this file.

// This file is part of DynExp.
 
#pragma once
 
#include "stdafx.h"
#include "Instrument.h"
 
namespace DynExpInstr
{
    class DataStreamInstrument;
 
    namespace DataStreamInstrumentTasks
    {
        class InitTask : public DynExp::InitTaskBase
        {
            void InitFuncImpl(dispatch_tag<InitTaskBase>, DynExp::InstrumentInstance& Instance) override final { InitFuncImpl(dispatch_tag<InitTask>(), Instance); }
 
            virtual void InitFuncImpl(dispatch_tag<InitTask>, DynExp::InstrumentInstance& Instance) {}
        };
 
        class ExitTask : public DynExp::ExitTaskBase
        {
            void ExitFuncImpl(dispatch_tag<ExitTaskBase>, DynExp::InstrumentInstance& Instance) override final { ExitFuncImpl(dispatch_tag<ExitTask>(), Instance); }
 
            virtual void ExitFuncImpl(dispatch_tag<ExitTask>, DynExp::InstrumentInstance& Instance) {}
        };
 
        class UpdateTask : public DynExp::UpdateTaskBase
        {
            void UpdateFuncImpl(dispatch_tag<UpdateTaskBase>, DynExp::InstrumentInstance& Instance) override final { UpdateFuncImpl(dispatch_tag<UpdateTask>(), Instance); }
 
            virtual void UpdateFuncImpl(dispatch_tag<UpdateTask>, DynExp::InstrumentInstance& Instance) {}
        };
 
        class SetStreamSizeTask final : public DynExp::TaskBase
        {
        public:
            SetStreamSizeTask(size_t BufferSizeInSamples, CallbackType CallbackFunc) noexcept : TaskBase(std::move(CallbackFunc)), BufferSizeInSamples(BufferSizeInSamples) {}
 
        private:
            virtual DynExp::TaskResultType RunChild(DynExp::InstrumentInstance& Instance) override;
 
            const size_t BufferSizeInSamples;   
        };
    }
 
    struct BasicSample
    {
        using DataType = double;    
 
        constexpr BasicSample() noexcept : Value(0), Time(0) {}
 
        constexpr BasicSample(DataType Value) noexcept : Value(Value), Time(0) {}
 
        constexpr BasicSample(DataType Value, DataType Time) noexcept : Value(Value), Time(Time) {}
 
        DataType Value;             
        DataType Time;              
    };
 
    class DataStreamBase
    {
    public:
        using BasicSampleListType = std::vector<BasicSample>;
 
        virtual ~DataStreamBase() = default;
 
 
        virtual bool IsBasicSampleConvertible() const noexcept { return false; }
 
        virtual bool IsBasicSampleTimeUsed() const noexcept { return false; }
 
        virtual void SeekBeg(std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) = 0;
 
        virtual void SeekEnd(std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) = 0;
 
        virtual bool SeekEqual(std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) = 0;
 
        virtual size_t GetStreamSizeRead() const noexcept = 0;
 
        virtual size_t GetStreamSizeWrite() const noexcept = 0;
 
        virtual size_t GetNumSamplesWritten() const noexcept = 0;
 
        virtual void SetStreamSize(size_t BufferSizeInSamples) = 0;
 
        bool CanRead() const { return GetStreamSizeRead(); }
 
        void Clear() { ClearChild(); }
 
        void WriteBasicSample(const BasicSample& Sample) { WriteBasicSampleChild(Sample); }
 
        BasicSample ReadBasicSample() { return ReadBasicSampleChild(); }
 
        void WriteBasicSamples(const BasicSampleListType& Samples);
 
        BasicSampleListType ReadBasicSamples(size_t Count);
 
    private:
 
        virtual void WriteBasicSampleChild(const BasicSample& Sample);
 
        virtual BasicSample ReadBasicSampleChild();
 
        virtual void ClearChild() = 0;      
    };
 
    template <typename T>
    using DataStreamPtrType = std::unique_ptr<T>;
 
    using DataStreamBasePtrType = DataStreamPtrType<DataStreamBase>;
 
    class CircularDataStreamBase : public DataStreamBase
    {
    protected:
        CircularDataStreamBase() = default;
        virtual ~CircularDataStreamBase() = default;
 
    public:
 
        virtual size_t GetNumAvailableSamplesToReadTillEnd() const noexcept = 0;
 
        virtual size_t GetNumFreeSamplesToWrite() const noexcept = 0;
 
        virtual std::streampos GetReadPosition() const noexcept = 0;
 
        virtual std::streampos GetWritePosition() const noexcept = 0;
 
        virtual bool SeekRel(signed long long OffsetInSamples, std::ios_base::seekdir SeekDir,
            std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) = 0;
 
        virtual bool SeekAbs(unsigned long long PositionInSamples, std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) = 0;
 
        size_t GetNumRecentBasicSamples(size_t Count) const;
 
        BasicSampleListType ReadRecentBasicSamples(size_t Count);
    };
 
    template <
        typename SampleT,
        std::enable_if_t<std::is_trivially_copyable_v<SampleT>, int> = 0
    >
    class CircularDataStream : public CircularDataStreamBase
    {
    public:
        using SampleType = SampleT;     
 
        CircularDataStream(size_t BufferSizeInSamples)
            : StreamBuffer(sizeof(SampleT) * BufferSizeInSamples), Stream(&StreamBuffer), NumSamplesWritten(0)
        {
            Stream.exceptions(std::iostream::failbit | std::iostream::badbit);
        }
 
        virtual ~CircularDataStream() = default;
 
        constexpr auto GetBytesPerSample() noexcept { return sizeof(SampleT); }
 
        size_t GetNumAvailableSamplesToReadTillEnd() const noexcept override { return (StreamBuffer.gsize() - StreamBuffer.gtellp()) / sizeof(SampleT); }
        size_t GetNumFreeSamplesToWrite() const noexcept override { return (StreamBuffer.psize() - StreamBuffer.ptellp()) / sizeof(SampleT); }
        std::streampos GetReadPosition() const noexcept override { return StreamBuffer.gtellp() / sizeof(SampleT); }
        std::streampos GetWritePosition() const noexcept override { return StreamBuffer.ptellp() / sizeof(SampleT); }
 
        virtual bool SeekRel(signed long long OffsetInSamples, std::ios_base::seekdir SeekDir,
            std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) override
        {
            return StreamBuffer.pubseekoff(OffsetInSamples * sizeof(SampleT), SeekDir, Which) !=
                Util::circularbuf::pos_type(Util::circularbuf::off_type(-1));
        }
 
        virtual bool SeekAbs(unsigned long long PositionInSamples,
            std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) override
        {
            return StreamBuffer.pubseekpos(PositionInSamples * sizeof(SampleT), Which) !=
                Util::circularbuf::pos_type(Util::circularbuf::off_type(-1));
        }
 
        virtual void SeekBeg(std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) override
        {
            SeekAbs(0, Which);
        }
 
        virtual void SeekEnd(std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) override
        {
            if (Which & std::ios_base::in)
                SeekRel(-1, std::ios_base::end, std::ios_base::in);
            if (Which & std::ios_base::out)
                SeekRel(0, std::ios_base::end, std::ios_base::out);
        }
 
        virtual bool SeekEqual(std::ios_base::openmode Which = std::ios_base::in | std::ios_base::out) override
        {
            if (Which & std::ios_base::in)
            {
                StreamBuffer.pubsync();
 
                if (GetStreamSizeRead() >= GetStreamSizeWrite())
                    SeekAbs(GetWritePosition(), std::ios_base::in);
                else
                    return false;
            }
 
            // Read buffer is never larger than write buffer.
            if (Which & std::ios_base::out)
                SeekAbs(GetReadPosition(), std::ios_base::out);
 
            return true;
        }
 
        virtual size_t GetStreamSizeRead() const noexcept override { return StreamBuffer.gsize() / sizeof(SampleT);}
        virtual size_t GetStreamSizeWrite() const noexcept override { return StreamBuffer.psize() / sizeof(SampleT); }
        virtual size_t GetNumSamplesWritten() const noexcept override { return NumSamplesWritten; }
        virtual void SetStreamSize(size_t BufferSizeInSamples) override { StreamBuffer.resize(sizeof(SampleT) * BufferSizeInSamples); }
 
 
        void WriteSample(const SampleT& Sample)
        {
            ValidateSample(Sample);
 
            Stream.write(reinterpret_cast<const char*>(&Sample), sizeof(SampleT));
            
            // Should never overflow. Let's assume 10 Gb/s = 1.25e9 B/s transmission rate:
            // Overflows in (2^64 - 1) B / 1.25e9 B/s / 60 / 60 / 24 / 365 = 468 years.
            if (NumSamplesWritten < std::numeric_limits<decltype(NumSamplesWritten)>::max())
                ++NumSamplesWritten;
        }
 
        template <typename Rep, typename Period>
        void WriteSample(const std::chrono::duration<Rep, Period>& Sample)
        {
            WriteSample(Sample.count());
        }
 
        SampleT ReadSample()
        {
            SampleT Sample;
            Stream.read(reinterpret_cast<char*>(&Sample), sizeof(SampleT));
 
            return Sample;
        }
 
        template <typename T>
        void WriteSamples(const std::vector<T>& Samples)
        {
            for (const auto& Sample : Samples)
                WriteSample(Sample);
        }
 
        std::vector<SampleT> ReadSamples(size_t Count)
        {
            std::vector<SampleT> Samples;
 
            for (decltype(Count) i = 0; i < Count; ++i)
                Samples.push_back(ReadSample());
 
            return Samples;
        }
 
    protected:
        virtual void ClearChild() override
        {
            Stream.clear();
            StreamBuffer.clear();
        }
 
    private:
 
        virtual void ValidateSample(const SampleT& Sample) const {}
 
        mutable Util::circularbuf StreamBuffer;     
        std::iostream Stream;                       
        size_t NumSamplesWritten;                   
    };
 
    class BasicSampleStream : public CircularDataStream<BasicSample>
    {
    public:
        BasicSampleStream(size_t BufferSizeInSamples) : CircularDataStream(BufferSizeInSamples) {}
 
        virtual ~BasicSampleStream() = default;
 
        bool IsBasicSampleConvertible() const noexcept override final { return true; }
        bool IsBasicSampleTimeUsed() const noexcept override final { return true; }
 
    private:
        virtual void WriteBasicSampleChild(const BasicSample& Sample) override { WriteSample(Sample); }
        virtual BasicSample ReadBasicSampleChild() override { return ReadSample(); }
    };
 
    template <
        typename SampleT,
        std::enable_if_t<std::is_arithmetic_v<SampleT>, int> = 0
    >
    class NumericSampleStream : public CircularDataStream<SampleT>
    {
    public:
        NumericSampleStream(size_t BufferSizeInSamples) : CircularDataStream<SampleT>(BufferSizeInSamples),
            MinValue(std::numeric_limits<SampleT>::lowest()), MaxValue(std::numeric_limits<SampleT>::max()) {}
        
        NumericSampleStream(size_t BufferSizeInSamples, SampleT MinValue, SampleT MaxValue)
            : CircularDataStream<SampleT>(BufferSizeInSamples), MinValue(MinValue), MaxValue(MaxValue) {}
        
        virtual ~NumericSampleStream() = default;
 
        bool IsBasicSampleConvertible() const noexcept override final { return true; }
 
        void SetLimits(SampleT MinValue, SampleT MaxValue)
        {
            this->MinValue = MinValue;
            this->MaxValue = MaxValue;
        }
 
        auto GetMinValue() const noexcept { return MinValue; }
 
        auto GetMaxValue() const noexcept { return MaxValue; }
 
    private:
        virtual void WriteBasicSampleChild(const BasicSample& Sample) override
        {
            CircularDataStream<SampleT>::WriteSample(static_cast<SampleT>(Sample.Value));
        }
 
        virtual BasicSample ReadBasicSampleChild() override
        {
            BasicSample Sample;
 
            // Ignore time value (set it to 0). It has been considered (and discarded) to set it to
            // std::chrono::system_clock::now() instead. This would be misleading since a timestamp
            // which is set here refers to the time when the sample is read from the buffer - not
            // to the time when it is actually acquired by the underlying hardware.
            Sample.Time = 0;
            Sample.Value = static_cast<decltype(Sample.Value)>(CircularDataStream<SampleT>::ReadSample());
 
            return Sample;
        }
 
        virtual void ValidateSample(const SampleT& Sample) const override
        {
            if (Sample < MinValue || Sample > MaxValue)
                throw Util::InvalidDataException(
                    "A sample cannot be written to a stream buffer because it exceeds the specified limits (min: " + std::to_string(MinValue)
                    + ", max: " + std::to_string(MaxValue) + ", sample: " + std::to_string(Sample) + ")");
        }
 
        SampleT MinValue;       
        SampleT MaxValue;       
    };
 
    using AnalogSampleStream = NumericSampleStream<double>;                     
    using DigitalSampleStream = NumericSampleStream<uint8_t>;                   
    using AnalogSampleStreamPtrType = DataStreamPtrType<AnalogSampleStream>;    
    using DigitalSampleStreamPtrType = DataStreamPtrType<DigitalSampleStream>;  
 
    class StreamSizeParamsExtension
    {
        static constexpr ParamsConfigDialog::NumberType DefaultStreamSize = 1000;   
 
    public:
        struct ValueType
        {
            ParamsConfigDialog::NumberType StreamSize = DefaultStreamSize;          
        };
 
        StreamSizeParamsExtension(DynExp::ParamsBase& Owner,
            ParamsConfigDialog::NumberType DefaultValue = DefaultStreamSize, ParamsConfigDialog::NumberType MinValue = 1,
            ParamsConfigDialog::NumberType MaxValue = static_cast<double>(std::numeric_limits<size_t>::max()))
            : StreamSize{ Owner, "StreamSize", "Stream size",
                "Size of the instrument's sample stream buffer in samples.", true, DefaultValue, MinValue, MaxValue, 1, 0 }
        {}
 
        void DisableUserEditable();
 
        ValueType Values() const { return { StreamSize }; }
 
        DynExp::ParamsBase::Param<ParamsConfigDialog::NumberType> StreamSize;
    };
 
    class NumericSampleStreamParamsExtension
    {
    public:
        enum SamplingModeType { Single, Continuous };
    private:
        static constexpr ParamsConfigDialog::NumberType DefaultSamplingRate = 1;            
        static constexpr SamplingModeType DefaultSamplingMode = SamplingModeType::Single;   
 
    public:
        struct ValueType
        {
            ParamsConfigDialog::NumberType SamplingRate = DefaultSamplingRate;              
            SamplingModeType SamplingMode = DefaultSamplingMode;                            
        };
 
        static Util::TextValueListType<SamplingModeType> SamplingModeTypeStrList();
 
        NumericSampleStreamParamsExtension(DynExp::ParamsBase& Owner)
            : SamplingRate{ Owner, "SamplingRate", "Sampling rate",
                "Sampling rate in samples per second.", true, DefaultSamplingRate, 0, std::numeric_limits<ParamsConfigDialog::NumberType>::max(), 1, 3 },
            SamplingMode{ Owner, SamplingModeTypeStrList(), "SamplingMode", "Sampling mode",
                "Determines whether reading/writing happens only once or continuously.", true, DefaultSamplingMode }
        {}
 
        void DisableUserEditable();
 
        ValueType Values() const { return { SamplingRate, SamplingMode}; }
 
        DynExp::ParamsBase::Param<ParamsConfigDialog::NumberType> SamplingRate;
 
        DynExp::ParamsBase::Param<SamplingModeType> SamplingMode;
    };
 
    class DataStreamInstrumentData : public DynExp::InstrumentDataBase
    {
    public:
        using ValueType = double;
 
        enum class UnitType {
            Arbitrary,      
            LogicLevel,     
            Counts,         
            Volt,           
            Ampere,         
            Power_W,        
            Power_dBm       
        };
 
        static const char* UnitTypeToStr(const UnitType& Unit);
 
        DataStreamInstrumentData(DataStreamBasePtrType&& SampleStream);
 
        virtual ~DataStreamInstrumentData() = default;
 
        DataStreamBasePtrType::element_type* GetSampleStream() const noexcept { return SampleStream.get(); }
 
        template <typename T>
        auto GetCastSampleStream() const
        {
            auto CastSampleStream = dynamic_cast<T*>(GetSampleStream());
            if (!CastSampleStream)
                throw Util::TypeErrorException();
 
            return CastSampleStream;
        }
 
        ValueType GetHardwareMinValue() const noexcept { return HardwareMinValue; }         
        ValueType GetHardwareMaxValue() const noexcept { return HardwareMaxValue; }         
        UnitType GetValueUnit() const noexcept { return ValueUnit; }                        
        void SetHardwareMinValue(ValueType Value) noexcept { HardwareMinValue = Value; }    
        void SetHardwareMaxValue(ValueType Value) noexcept { HardwareMaxValue = Value; }    
        void SetValueUnit(UnitType Unit) noexcept { ValueUnit = Unit; }                     
 
    private:
        void ResetImpl(dispatch_tag<InstrumentDataBase>) override final;
        virtual void ResetImpl(dispatch_tag<DataStreamInstrumentData>) {};  
 
        const DataStreamBasePtrType SampleStream;                           
 
        ValueType HardwareMinValue;     
        ValueType HardwareMaxValue;     
        UnitType ValueUnit;             
    };
 
    class DataStreamInstrumentParams : public DynExp::InstrumentParamsBase
    {
    public:
        DataStreamInstrumentParams(DynExp::ItemIDType ID, const DynExp::DynExpCore& Core) : InstrumentParamsBase(ID, Core) {}
 
        virtual ~DataStreamInstrumentParams() = 0;
 
        virtual const char* GetParamClassTag() const noexcept override { return "DataStreamInstrumentParams"; }
 
    private:
        void ConfigureParamsImpl(dispatch_tag<InstrumentParamsBase>) override final { ConfigureParamsImpl(dispatch_tag<DataStreamInstrumentParams>()); }
        virtual void ConfigureParamsImpl(dispatch_tag<DataStreamInstrumentParams>) {}   
 
        DummyParam Dummy = { *this };                                                   
    };
 
    class DataStreamInstrumentConfigurator : public DynExp::InstrumentConfiguratorBase
    {
    public:
        using ObjectType = DataStreamInstrument;
        using ParamsType = DataStreamInstrumentParams;
 
        DataStreamInstrumentConfigurator() = default;
        virtual ~DataStreamInstrumentConfigurator() = 0;
    };
 
    class DataStreamInstrument : public DynExp::InstrumentBase
    {
    public:
        using ParamsType = DataStreamInstrumentParams;                              
        using ConfigType = DataStreamInstrumentConfigurator;                        
        using InstrumentDataType = DataStreamInstrumentData;                        
 
        using InitTaskType = DataStreamInstrumentTasks::InitTask;                   
        using ExitTaskType = DataStreamInstrumentTasks::ExitTask;                   
        using UpdateTaskType = DataStreamInstrumentTasks::UpdateTask;               
 
        constexpr static auto Name() noexcept { return "Data Stream Instrument"; }  
        constexpr static auto Category() noexcept { return "I/O"; }                 
 
        DataStreamInstrument(const std::thread::id OwnerThreadID, DynExp::ParamsBasePtrType&& Params)
            : InstrumentBase(OwnerThreadID, std::move(Params)) {}
 
        virtual ~DataStreamInstrument() = 0;
 
        virtual std::string GetName() const override { return Name(); }
        virtual std::string GetCategory() const override { return Category(); }
 
 
        virtual DataStreamInstrumentData::UnitType GetValueUnit() const = 0;
 
        const char* GetValueUnitStr() const noexcept { return DataStreamInstrumentData::UnitTypeToStr(GetValueUnit()); }
 
 
        virtual void ReadData(DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const;
 
        virtual void WriteData(DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const;
 
        virtual void ClearData(DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const {}
 
        virtual void Start(DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const {}
 
        virtual void Stop(DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const {}
 
        virtual void Restart(DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const;
 
        virtual void SetStreamSize(size_t BufferSizeInSamples, DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const;
 
        virtual void ResetStreamSize(DynExp::TaskBase::CallbackType CallbackFunc = nullptr) const {}
 
        virtual Util::OptionalBool HasFinished() const { return Util::OptionalBool::Values::Unknown; }
 
        virtual Util::OptionalBool IsRunning() const { return Util::OptionalBool::Values::Unknown; }
 
        bool CanRead(const std::chrono::milliseconds Timeout = GetInstrumentDataTimeoutDefault) const;
 
        void Clear(const std::chrono::milliseconds Timeout = GetInstrumentDataTimeoutDefault) const;
 
    private:
        void ResetImpl(dispatch_tag<InstrumentBase>) override final;
        virtual void ResetImpl(dispatch_tag<DataStreamInstrument>) = 0;     
 
        virtual std::unique_ptr<DynExp::InitTaskBase> MakeInitTask() const override { return DynExp::MakeTask<DataStreamInstrumentTasks::InitTask>(); }
        virtual std::unique_ptr<DynExp::ExitTaskBase> MakeExitTask() const override { return DynExp::MakeTask<DataStreamInstrumentTasks::ExitTask>(); }
        virtual std::unique_ptr<DynExp::UpdateTaskBase> MakeUpdateTask() const override { return DynExp::MakeTask<DataStreamInstrumentTasks::UpdateTask>(); }
    };
}