StreamManipulator.cpp

Go to the documentation of this file.

// This file is part of DynExp.
 
#include "stdafx.h"
#include "StreamManipulator.h"
 
PYBIND11_MAKE_OPAQUE(DynExpModule::PyStreamListType);
PYBIND11_MAKE_OPAQUE(decltype(DynExpModule::PyStreamManipulatorOutputData::LastConsumedSampleIDsPerInputStream));
 
PYBIND11_EMBEDDED_MODULE(PyModuleStreamManipulator, m)
{
    using namespace DynExpModule;
 
    py::bind_vector<DynExpModule::PyStreamListType>(m, "StreamListType");
    py::bind_vector<decltype(DynExpModule::PyStreamManipulatorOutputData::LastConsumedSampleIDsPerInputStream)>(m, "SampleIDListType");
 
    py::class_<PyStreamManipulatorInputData>(m, "InputData")
        .def(py::init<>())
        .def_readonly("ModuleID", &PyStreamManipulatorInputData::ModuleID)
        .def_readonly("LastExecutionTime", &PyStreamManipulatorInputData::LastExecutionTime)
        .def_readwrite("InputStreams", &PyStreamManipulatorInputData::InputStreams)
        .def_readwrite("OutputStreams", &PyStreamManipulatorInputData::OutputStreams)
        .def_readonly("SaveFilename",  & PyStreamManipulatorInputData::SaveFilename);
 
    py::class_<PyStreamManipulatorOutputData>(m, "OutputData")
        .def(py::init<>())
        .def_readwrite("MinNextExecutionDelay", &PyStreamManipulatorOutputData::MinNextExecutionDelay)
        .def_readwrite("MaxNextExecutionDelay", &PyStreamManipulatorOutputData::MaxNextExecutionDelay)
        .def_readwrite("LastConsumedSampleIDsPerInputStream", &PyStreamManipulatorOutputData::LastConsumedSampleIDsPerInputStream);
}
 
namespace DynExpModule
{
    void PyStreamManipulatorInputData::Reset()
    {
        ModuleID = DynExp::ItemIDNotSet;
        LastExecutionTime = {};
 
        InputStreams.clear();
        OutputStreams.clear();
 
        SaveFilename.clear();
    }
 
    void PyStreamManipulatorOutputData::Reset()
    {
        MinNextExecutionDelay = {};
        MaxNextExecutionDelay = {};
 
        LastConsumedSampleIDsPerInputStream.clear();
    }
 
    void StreamManipulatorData::ResetImpl(dispatch_tag<ModuleDataBase>)
    {
        Init();
    }
 
    void StreamManipulatorData::Init()
    {
    }
 
    Util::DynExpErrorCodes::DynExpErrorCodes StreamManipulator::ModuleMainLoop(DynExp::ModuleInstance& Instance)
    {
        try
        {
            auto ModuleData = DynExp::dynamic_ModuleData_cast<StreamManipulator>(Instance.ModuleDataGetter());
 
            bool IsNewDataAvlbl = false;
            for (size_t i = 0; i < ManipulatorPyFuncInput.InputStreams.size(); ++i)
            {
                auto& Instrument = ModuleData->GetInputDataStreams()[i];
                auto InstrData = DynExp::dynamic_InstrumentData_cast<DynExpInstr::DataStreamInstrument>(Instrument->GetInstrumentData());
                auto SampleStream = InstrData->GetCastSampleStream<SampleStreamType>();
 
                Instrument->ReadData();
 
                if (ManipulatorPyFuncOutput.LastConsumedSampleIDsPerInputStream.size() > i &&
                    SampleStream->GetNumRecentBasicSamples(ManipulatorPyFuncOutput.LastConsumedSampleIDsPerInputStream[i]))
                    IsNewDataAvlbl = true;
            }
 
            const auto now = std::chrono::system_clock::now();
            if (ManipulatorPyFuncStep &&
                ((ManipulatorPyFuncOutput.MaxNextExecutionDelay.count() && now - LastManipulatorPyFuncExecution >= ManipulatorPyFuncOutput.MaxNextExecutionDelay) ||
                (IsNewDataAvlbl && now - LastManipulatorPyFuncExecution >= ManipulatorPyFuncOutput.MinNextExecutionDelay) ||
                !LastManipulatorPyFuncExecution.time_since_epoch().count()))
                Step(ModuleData, ManipulatorPyFuncStep);
 
            NumFailedUpdateAttempts = 0;
        } // ModuleData and instruments' data unlocked here.
        catch (const Util::TimeoutException& e)
        {
            if (NumFailedUpdateAttempts++ >= 3)
                Instance.GetOwner().SetWarning(e);
        }
 
        return Util::DynExpErrorCodes::NoError;
    }
 
    void StreamManipulator::ResetImpl(dispatch_tag<ModuleBase>)
    {
        IsInitialized = false;
 
        ManipulatorPyFuncName = "StreamManipFunc_" + Util::ToStr(GetID());
        ManipulatorPyFuncInit.Reset();
        ManipulatorPyFuncStep.Reset();
        ManipulatorPyFuncExit.Reset();
        ManipulatorPyFuncFinished.Reset();
        ManipulatorPyFuncStart.Reset();
        ManipulatorPyFuncStop.Reset();
        ManipulatorPyFuncTrigger.Reset();
        ManipulatorPyFuncInput.Reset();
        ManipulatorPyFuncOutput.Reset();
 
        NumFailedUpdateAttempts = 0;
        LastManipulatorPyFuncExecution = {};
    }
 
    void StreamManipulator::Step(Util::SynchronizedPointer<ModuleDataType>& ModuleData, const PyFuncType& StepFunction, bool UpdateLastExecutionTime) const
    {
        for (size_t i = 0; i < ManipulatorPyFuncInput.InputStreams.size(); ++i)
        {
            auto& Instrument = ModuleData->GetInputDataStreams()[i];
            auto InstrData = DynExp::dynamic_InstrumentData_cast<DynExpInstr::DataStreamInstrument>(Instrument->GetInstrumentData());
            auto SampleStream = InstrData->GetCastSampleStream<SampleStreamType>();
 
            ManipulatorPyFuncInput.InputStreams[i].StreamSizeRead = SampleStream->GetStreamSizeRead();
            ManipulatorPyFuncInput.InputStreams[i].StreamSizeWrite = SampleStream->GetStreamSizeWrite();
            ManipulatorPyFuncInput.InputStreams[i].NumSamplesWritten = SampleStream->GetNumSamplesWritten();
            ManipulatorPyFuncInput.InputStreams[i].Samples = SampleStream->ReadRecentBasicSamples(ManipulatorPyFuncOutput.LastConsumedSampleIDsPerInputStream[i]);
        }
        for (size_t i = 0; i < ManipulatorPyFuncInput.OutputStreams.size(); ++i)
        {
            auto& Instrument = ModuleData->GetOutputDataStreams()[i];
            auto InstrData = DynExp::dynamic_InstrumentData_cast<DynExpInstr::DataStreamInstrument>(Instrument->GetInstrumentData());
            auto SampleStream = InstrData->GetCastSampleStream<SampleStreamType>();
 
            ManipulatorPyFuncInput.OutputStreams[i].StreamSizeRead = SampleStream->GetStreamSizeRead();
            ManipulatorPyFuncInput.OutputStreams[i].StreamSizeWrite = SampleStream->GetStreamSizeWrite();
            ManipulatorPyFuncInput.OutputStreams[i].NumSamplesWritten = SampleStream->GetNumSamplesWritten();
            ManipulatorPyFuncInput.OutputStreams[i].Samples.clear();
        }
 
        ManipulatorPyFuncInput.LastExecutionTime = LastManipulatorPyFuncExecution;
 
        PyStreamManipulatorOutputData FuncOutput;
        {
            py::gil_scoped_acquire acquire;
 
            auto PyResult = StepFunction(&ManipulatorPyFuncInput);
            if (!PyResult.is_none())
                FuncOutput = PyResult.cast<PyStreamManipulatorOutputData>();
        } // GIL released here.
        
        if (UpdateLastExecutionTime)
            LastManipulatorPyFuncExecution = std::chrono::system_clock::now();
        ManipulatorPyFuncOutput.MinNextExecutionDelay = FuncOutput.MinNextExecutionDelay;
        ManipulatorPyFuncOutput.MaxNextExecutionDelay = FuncOutput.MaxNextExecutionDelay;
 
        for (size_t i = 0; i < ManipulatorPyFuncInput.InputStreams.size(); ++i)
        {
            auto& Instrument = ModuleData->GetInputDataStreams()[i];
            auto InstrData = DynExp::dynamic_InstrumentData_cast<DynExpInstr::DataStreamInstrument>(Instrument->GetInstrumentData());
            auto SampleStream = InstrData->GetCastSampleStream<SampleStreamType>();
 
            if (FuncOutput.LastConsumedSampleIDsPerInputStream.size() > i
                && FuncOutput.LastConsumedSampleIDsPerInputStream[i] < SampleStream->GetNumSamplesWritten()
                && FuncOutput.LastConsumedSampleIDsPerInputStream[i] >= SampleStream->GetNumSamplesWritten() - SampleStream->GetStreamSizeRead())
                ManipulatorPyFuncOutput.LastConsumedSampleIDsPerInputStream[i] = FuncOutput.LastConsumedSampleIDsPerInputStream[i];
            else
                ManipulatorPyFuncOutput.LastConsumedSampleIDsPerInputStream[i] = SampleStream->GetNumSamplesWritten();
 
            if (ManipulatorPyFuncInput.InputStreams[i].ShouldCLear())
            {
                SampleStream->Clear();
                ManipulatorPyFuncInput.InputStreams[i].Cleared();
            }
        }
        for (size_t i = 0; i < ManipulatorPyFuncInput.OutputStreams.size(); ++i)
        {
            auto& Instrument = ModuleData->GetOutputDataStreams()[i];
            auto InstrData = DynExp::dynamic_InstrumentData_cast<DynExpInstr::DataStreamInstrument>(Instrument->GetInstrumentData());
            auto SampleStream = InstrData->GetCastSampleStream<SampleStreamType>();
 
            if (ManipulatorPyFuncInput.OutputStreams[i].ShouldCLear())
            {
                SampleStream->Clear();
                ManipulatorPyFuncInput.OutputStreams[i].Cleared();
            }
 
            if (!ManipulatorPyFuncInput.OutputStreams[i].Samples.empty())
            {
                SampleStream->WriteBasicSamples(ManipulatorPyFuncInput.OutputStreams[i].Samples);
                Instrument->WriteData();
            }
        }
    }
 
    void StreamManipulator::OnInit(DynExp::ModuleInstance* Instance) const
    {
        SetFilenameEvent::Register(*this, &StreamManipulator::OnSetFilename);
 
        auto ModuleParams = DynExp::dynamic_Params_cast<StreamManipulator>(Instance->ParamsGetter());
        auto ModuleData = DynExp::dynamic_ModuleData_cast<StreamManipulator>(Instance->ModuleDataGetter());
 
        Instance->LockObject(ModuleParams->InputDataStreams, ModuleData->GetInputDataStreams());
        Instance->LockObject(ModuleParams->OutputDataStreams, ModuleData->GetOutputDataStreams());
 
        if (ModuleParams->Communicator.ContainsID())
            Instance->LockObject(ModuleParams->Communicator, ModuleData->GetCommunicator());
 
        auto PythonCode = Util::ReadFromFile(ModuleParams->PythonCodePath.GetPath());
        PythonCode = std::regex_replace(PythonCode, std::regex("\r\n"), "\n");
        PythonCode = std::regex_replace(PythonCode, std::regex("\n"), std::string("\n") + Util::PyTab);
 
        py::gil_scoped_acquire acquire;
        DynExpInstr::PyDataStreamInstrument::import();
        py::exec("import PyModuleStreamManipulator as StreamManipulator");
        py::exec("def " + ManipulatorPyFuncName + "():\n" +
            Util::PyTab + PythonCode + "\n" +
            Util::PyTab + "if 'on_init' in locals() and callable(on_init):\n" +
            Util::PyTab + Util::PyTab + ManipulatorPyFuncName + ".init = on_init\n" +
            Util::PyTab + "if 'on_step' in locals() and callable(on_step):\n" +
            Util::PyTab + Util::PyTab + ManipulatorPyFuncName + ".step = on_step\n" +
            Util::PyTab + "if 'on_exit' in locals() and callable(on_exit):\n" +
            Util::PyTab + Util::PyTab + ManipulatorPyFuncName + ".exit = on_exit\n" +
            Util::PyTab + "if 'on_finished' in locals() and callable(on_finished):\n" +
            Util::PyTab + Util::PyTab + ManipulatorPyFuncName + ".finished = on_finished\n" +
            Util::PyTab + "if 'on_start' in locals() and callable(on_start):\n" +
            Util::PyTab + Util::PyTab + ManipulatorPyFuncName + ".start = on_start\n" +
            Util::PyTab + "if 'on_stop' in locals() and callable(on_stop):\n" +
            Util::PyTab + Util::PyTab + ManipulatorPyFuncName + ".stop = on_stop\n" +
            Util::PyTab + "if 'on_trigger' in locals() and callable(on_trigger):\n" +
            Util::PyTab + Util::PyTab + ManipulatorPyFuncName + ".trigger = on_trigger");
        auto ManipulatorPyFunc = py::eval(ManipulatorPyFuncName);
        ManipulatorPyFunc();
        ManipulatorPyFuncInit = py::hasattr(ManipulatorPyFunc, "init") ? py::getattr(ManipulatorPyFunc, "init") : py::none();
        ManipulatorPyFuncStep = py::hasattr(ManipulatorPyFunc, "step") ? py::getattr(ManipulatorPyFunc, "step") : py::none();
        ManipulatorPyFuncExit = py::hasattr(ManipulatorPyFunc, "exit") ? py::getattr(ManipulatorPyFunc, "exit") : py::none();
        ManipulatorPyFuncFinished = py::hasattr(ManipulatorPyFunc, "finished") ? py::getattr(ManipulatorPyFunc, "finished") : py::none();
        ManipulatorPyFuncStart = py::hasattr(ManipulatorPyFunc, "start") ? py::getattr(ManipulatorPyFunc, "start") : py::none();
        ManipulatorPyFuncStop = py::hasattr(ManipulatorPyFunc, "stop") ? py::getattr(ManipulatorPyFunc, "stop") : py::none();
        ManipulatorPyFuncTrigger = py::hasattr(ManipulatorPyFunc, "trigger") ? py::getattr(ManipulatorPyFunc, "trigger") : py::none();
 
        if (ModuleParams->Communicator.ContainsID() && ManipulatorPyFuncFinished)
            FinishedEvent::Register(*this, &StreamManipulator::OnFinished);
        if (ModuleParams->Communicator.ContainsID() && ManipulatorPyFuncStart)
            StartEvent::Register(*this, &StreamManipulator::OnStart);
        if (ModuleParams->Communicator.ContainsID() && ManipulatorPyFuncStop)
            StopEvent::Register(*this, &StreamManipulator::OnStop);
        if (ModuleParams->Communicator.ContainsID() && ManipulatorPyFuncTrigger)
            TriggerEvent::Register(*this, &StreamManipulator::OnTrigger);
 
        ManipulatorPyFuncInput.ModuleID = GetID();
        for (size_t i = 0; i < ModuleData->GetInputDataStreams().GetList().size(); ++i)
        {
            auto& Instrument = ModuleData->GetInputDataStreams()[i];
            auto InstrData = DynExp::dynamic_InstrumentData_cast<DynExpInstr::DataStreamInstrument>(Instrument->GetInstrumentData());
            auto SampleStream = InstrData->GetCastSampleStream<SampleStreamType>();
 
            ManipulatorPyFuncInput.InputStreams.emplace_back(SampleStream->IsBasicSampleTimeUsed(), Instrument->GetValueUnit());
            ManipulatorPyFuncOutput.LastConsumedSampleIDsPerInputStream.push_back(0);
        }
        for (size_t i = 0; i < ModuleData->GetOutputDataStreams().GetList().size(); ++i)
        {
            auto& Instrument = ModuleData->GetOutputDataStreams()[i];
            auto InstrData = DynExp::dynamic_InstrumentData_cast<DynExpInstr::DataStreamInstrument>(Instrument->GetInstrumentData());
            auto SampleStream = InstrData->GetCastSampleStream<SampleStreamType>();
 
            ManipulatorPyFuncInput.OutputStreams.emplace_back(SampleStream->IsBasicSampleTimeUsed(), Instrument->GetValueUnit());
        }
 
        ManipulatorPyFuncInit(&ManipulatorPyFuncInput);
 
        IsInitialized = true;
    }
 
    void StreamManipulator::OnExit(DynExp::ModuleInstance* Instance) const
    {
        try 
        {
            py::gil_scoped_acquire acquire;
            ManipulatorPyFuncExit(&ManipulatorPyFuncInput);
            py::exec("del " + ManipulatorPyFuncName);
        }
        catch (...)
        {
            // Swallow any exception which might arise from the shutdown of the module's Python part
            // since a failure of that is not considered a severe error.
            Util::EventLog().Log("[StreamManipulator] Shutting down Python part of module \"" + GetObjectName() + "\" failed.", Util::ErrorType::Warning);
        }
 
        auto ModuleData = DynExp::dynamic_ModuleData_cast<StreamManipulator>(Instance->ModuleDataGetter());
 
        Instance->UnlockObject(ModuleData->GetInputDataStreams());
        Instance->UnlockObject(ModuleData->GetOutputDataStreams());
        Instance->UnlockObject(ModuleData->GetCommunicator());
 
        SetFilenameEvent::Deregister(*this);
        FinishedEvent::Deregister(*this);
        StartEvent::Deregister(*this);
        StopEvent::Deregister(*this);
        TriggerEvent::Deregister(*this);
    }
 
    void StreamManipulator::OnSetFilename(DynExp::ModuleInstance* Instance, const std::string& SaveFilename) const
    {
        // Add file extension on Python side.
        ManipulatorPyFuncInput.SaveFilename = SaveFilename;
    }
 
    void StreamManipulator::OnFinished(DynExp::ModuleInstance* Instance) const
    {
        auto ModuleData = DynExp::dynamic_ModuleData_cast<StreamManipulator>(Instance->ModuleDataGetter());
 
        Step(ModuleData, ManipulatorPyFuncFinished, false);
    }
 
    void StreamManipulator::OnStart(DynExp::ModuleInstance* Instance) const
    {
        auto ModuleData = DynExp::dynamic_ModuleData_cast<StreamManipulator>(Instance->ModuleDataGetter());
 
        Step(ModuleData, ManipulatorPyFuncStart, false);
    }
 
    void StreamManipulator::OnStop(DynExp::ModuleInstance* Instance) const
    {
        auto ModuleData = DynExp::dynamic_ModuleData_cast<StreamManipulator>(Instance->ModuleDataGetter());
 
        Step(ModuleData, ManipulatorPyFuncStop, false);
    }
 
    void StreamManipulator::OnTrigger(DynExp::ModuleInstance* Instance) const
    {
        auto ModuleData = DynExp::dynamic_ModuleData_cast<StreamManipulator>(Instance->ModuleDataGetter());
 
        Step(ModuleData, ManipulatorPyFuncTrigger, false);
 
        if (ModuleData->GetCommunicator().valid())
            ModuleData->GetCommunicator()->PostEvent(*this, FinishedEvent{});
    }
}