StreamManipulator.h

Go to the documentation of this file.

// This file is part of DynExp.
 
#pragma once
 
#include "stdafx.h"
#include "DynExpCore.h"
#include "PyModules.h"
#include "../Instruments/InterModuleCommunicator.h"
 
#include "CommonModuleEvents.h"
 
namespace DynExpModule
{
    class StreamManipulator;
 
    using PyStreamListType = std::vector<DynExpInstr::PyDataStreamInstrument>;
 
    struct PyStreamManipulatorInputData
    {
        DynExp::ItemIDType ModuleID{ DynExp::ItemIDNotSet };
 
        std::chrono::time_point<std::chrono::system_clock> LastExecutionTime{};
 
        PyStreamListType InputStreams;
 
        PyStreamListType OutputStreams;
 
        std::string SaveFilename;
 
        void Reset();
    };
 
    struct PyStreamManipulatorOutputData
    {
        std::chrono::milliseconds MinNextExecutionDelay{};
 
        std::chrono::milliseconds MaxNextExecutionDelay{};
 
        std::vector<size_t> LastConsumedSampleIDsPerInputStream;
 
        void Reset();
    };
 
    class StreamManipulatorData : public DynExp::ModuleDataBase
    {
    public:
        StreamManipulatorData() { Init(); }
        virtual ~StreamManipulatorData() = default;
 
        auto& GetInputDataStreams() const noexcept { return InputDataStreams; }     
        auto& GetInputDataStreams() noexcept { return InputDataStreams; }           
        auto& GetOutputDataStreams() const noexcept { return OutputDataStreams; }   
        auto& GetOutputDataStreams() noexcept { return OutputDataStreams; }         
        auto& GetCommunicator() noexcept { return Communicator; }                   
 
    private:
        void ResetImpl(dispatch_tag<ModuleDataBase>) override final;
 
        virtual void ResetImpl(dispatch_tag<StreamManipulatorData>) {};
 
        void Init();
 
        DynExp::LinkedObjectWrapperContainerList<DynExpInstr::DataStreamInstrument> InputDataStreams;
 
        DynExp::LinkedObjectWrapperContainerList<DynExpInstr::DataStreamInstrument> OutputDataStreams;
 
        DynExp::LinkedObjectWrapperContainer<DynExpInstr::InterModuleCommunicator> Communicator;
    };
 
    class StreamManipulatorParams : public DynExp::ModuleParamsBase
    {
    public:
        StreamManipulatorParams(DynExp::ItemIDType ID, const DynExp::DynExpCore& Core) : ModuleParamsBase(ID, Core) {}
 
        virtual ~StreamManipulatorParams() = default;
 
        virtual const char* GetParamClassTag() const noexcept override { return "StreamManipulatorParams"; }
 
        ListParam<DynExp::ObjectLink<DynExpInstr::DataStreamInstrument>> InputDataStreams = { *this, GetCore().GetInstrumentManager(),
            "InputDataStreams", "Input data stream instrument(s)", "Data stream instruments to be used as data sources", DynExpUI::Icons::Instrument, true };
        
        ListParam<DynExp::ObjectLink<DynExpInstr::DataStreamInstrument>> OutputDataStreams = { *this, GetCore().GetInstrumentManager(),
            "OutputDataStreams", "Output data stream instrument(s)", "Data stream instruments to be used as data sinks", DynExpUI::Icons::Instrument, true };
        
        Param<ParamsConfigDialog::TextType> PythonCodePath = { *this, "PythonCodePath", "Python code path",
            "Path to a Python file containing the function which writes samples to the output streams based on the input streams' data", true, "", DynExp::TextUsageType::Code };
 
        Param<DynExp::ObjectLink<DynExpInstr::InterModuleCommunicator>> Communicator = { *this, GetCore().GetInstrumentManager(),
            "InterModuleCommunicator", "Inter-module communicator", "Inter-module communicator to control this module with", DynExpUI::Icons::Instrument, true };
 
    private:
        void ConfigureParamsImpl(dispatch_tag<ModuleParamsBase>) override final {}
    };
 
    class StreamManipulatorConfigurator : public DynExp::ModuleConfiguratorBase
    {
    public:
        using ObjectType = StreamManipulator;
        using ParamsType = StreamManipulatorParams;
 
        StreamManipulatorConfigurator() = default;
        virtual ~StreamManipulatorConfigurator() = default;
 
    private:
        virtual DynExp::ParamsBasePtrType MakeParams(DynExp::ItemIDType ID, const DynExp::DynExpCore& Core) const override final { return DynExp::MakeParams<StreamManipulatorConfigurator>(ID, Core); }
    };
 
    class StreamManipulator : public DynExp::ModuleBase
    {
        using SampleStreamType = DynExpInstr::CircularDataStreamBase;
 
        using PyFuncType = Util::PyObject<py::object>;
 
    public:
        using ParamsType = StreamManipulatorParams;                             
        using ConfigType = StreamManipulatorConfigurator;                       
        using ModuleDataType = StreamManipulatorData;                           
 
        constexpr static auto Name() noexcept { return "Stream Manipulator"; }  
        constexpr static auto Category() noexcept { return "I/O"; }             
 
        StreamManipulator(const std::thread::id OwnerThreadID, DynExp::ParamsBasePtrType&& Params)
            : ModuleBase(OwnerThreadID, std::move(Params)) {}
 
        virtual ~StreamManipulator() = default;
 
        virtual std::string GetName() const override { return Name(); }
        virtual std::string GetCategory() const override { return Category(); }
 
        bool TreatModuleExceptionsAsWarnings() const override { return IsInitialized; }
 
        std::chrono::milliseconds GetMainLoopDelay() const override final { return std::chrono::milliseconds(1); }
 
    private:
        Util::DynExpErrorCodes::DynExpErrorCodes ModuleMainLoop(DynExp::ModuleInstance& Instance) override final;
 
        void ResetImpl(dispatch_tag<ModuleBase>) override final;
 
        void Step(Util::SynchronizedPointer<ModuleDataType>& ModuleData, const PyFuncType& StepFunction, bool UpdateLastExecutionTime = true) const;
 
        void OnInit(DynExp::ModuleInstance* Instance) const override final;
        void OnExit(DynExp::ModuleInstance* Instance) const override final;
 
        void OnSetFilename(DynExp::ModuleInstance* Instance, const std::string& Filename) const;
 
        void OnFinished(DynExp::ModuleInstance* Instance) const;
 
        void OnStart(DynExp::ModuleInstance* Instance) const;
 
        void OnStop(DynExp::ModuleInstance* Instance) const;
 
        void OnTrigger(DynExp::ModuleInstance* Instance) const;
 
        mutable std::atomic<bool> IsInitialized = false;
 
        std::string ManipulatorPyFuncName;
 
        mutable PyFuncType ManipulatorPyFuncInit;       
        mutable PyFuncType ManipulatorPyFuncStep;       
        mutable PyFuncType ManipulatorPyFuncExit;       
        mutable PyFuncType ManipulatorPyFuncFinished;   
        mutable PyFuncType ManipulatorPyFuncStart;      
        mutable PyFuncType ManipulatorPyFuncStop;       
        mutable PyFuncType ManipulatorPyFuncTrigger;    
 
        mutable PyStreamManipulatorInputData ManipulatorPyFuncInput;
 
        mutable PyStreamManipulatorOutputData ManipulatorPyFuncOutput;
 
        size_t NumFailedUpdateAttempts = 0;
 
        mutable std::chrono::time_point<std::chrono::system_clock> LastManipulatorPyFuncExecution{};
    };
}