mdd/lib/src/ProcessorStandard.cpp

#include "ProcessorStandard.h"

namespace mdd {
    ProcessorStandard::ProcessorStandard()
        :ProcessorBase(R"JSON(
        [{
            "name":"priority",
            "value":"manual",
            "options":  [
                            "manual",
                            "static",
                            "dynamic",
                            "time"                     
                        ]
        },{
            "name":"iterations",
            "value": -1
        }
        ])JSON")
    {
        _priorityEvaluation = MANUAL;
        _maxIterations = -1;
        
        key = "StandardProcessor";
        setName(key);
        processor_outputs.push_back(std::make_shared<Output>(this, "Iterator", 0, std::vector<double>{0}));
    }

    bool ProcessorStandard::configure(const std::string& config) {
        json config_parsed = json::parse(config);
        bool found = false;
        for (size_t i = 0; i < config_parsed.size(); i++)
        {
            if (config_parsed[i].contains("name"))
            {
                if (config_parsed[i]["name"].get<std::string>() == "priority")
                {
                    std::string op = config_parsed[i]["value"].get<std::string>();
                    found = true;
                    if (op == "manual") {
                        _priorityEvaluation = MANUAL;
                        break;
                    }
                    else if (op == "static") {
                        _priorityEvaluation = STATIC;
                    }
                    else if (op == "dynamic") {
                        _priorityEvaluation = DYNAMIC;
                    }
                    else if (op == "time") {
                        _priorityEvaluation = TIME;
                    }
                    else {
                        found = false;
                    }
                }
                if (config_parsed[i]["name"].get<std::string>() == "iterations")
                {
                    _maxIterations = config_parsed[i]["value"].get<int>();
                    found = true;
                }
            }
        }

        return found;
    }

    std::string ProcessorStandard::addModule(std::shared_ptr<IModule> module)
    {
        std::string id = ProcessorBase::addModule(module);
        //_priority_list.emplace_back(module);
        return id;
    }

    void ProcessorStandard::removeModule(std::shared_ptr<IModule> module)
    {
        ProcessorBase::removeModule(module);
        //for (auto it = _priority_list.begin(); it != _priority_list.end(); ++it) {
        //    if (it->module_ptr == module)
        //    {
        //        _priority_list.erase(it);
        //        return;
        //    }
        //}
    }

    std::vector<std::shared_ptr<IModule>> ProcessorStandard::getModulePriority()
    {
        std::vector <std::shared_ptr<IModule>> ret;
        for (auto& p : _priority_list) {
            ret.push_back(p.module_ptr);
        }
        return ret;
    }

    state ProcessorStandard::update() {
        _priority_list.clear();
        for (size_t i = 0; i < inputs.size(); i++)
        {
            _priority_list.emplace_back(inputs[i]);
        }

        for (size_t i = 0; i < modules.size(); i++)
        {
            _priority_list.emplace_back(modules[i]);
        }

        for (size_t i = 0; i < outputs.size(); i++)
        {
            _priority_list.emplace_back(outputs[i]);
        }

        if (_priorityEvaluation != MANUAL)
        {
            //update priorities
            for (auto it = _priority_list.begin(); it != _priority_list.end(); ++it) {
                //collect connected inputs
                it->inputCounter = 0;
                for (size_t i = 0; i < it->module_ptr->getNumInputs(); ++i) {
                    if (it->module_ptr->getInput(i)->getConnection() != nullptr)
                    {
                        ++(it->inputCounter);
                    }
                }
                //collect connected outputs
                for (size_t i = 0; i < it->module_ptr->getNumOutputs(); ++i) {
                    if (!it->module_ptr->getOutput(i)->getConnections().empty())
                    {
                        ++(it->outputCounter);
                    }
                }
            }
            //sort by connections
            std::sort(_priority_list.begin(), _priority_list.end(), [](module_priority a, module_priority b) {
                if (a.inputCounter == b.inputCounter)
                {
                    return a.outputCounter > b.outputCounter;
                }
                return a.inputCounter < b.inputCounter;
            });
        }
        
        typedef std::chrono::high_resolution_clock Time;
        //update
        state ret = state::UNCHANGED;
        state group_state = state::CHANGED;
        size_t restart = 0;
        while (group_state == state::CHANGED) {
            processor_outputs[0]->setValue()[0] += 1;
            processor_outputs[0]->resetState();
            group_state = state::UNCHANGED;
            for (int i = restart; i < _priority_list.size(); ++i) {
                auto t_start = Time::now();
                state module_state = _priority_list[i].module_ptr->update();
                auto t_end = Time::now();
                if (_priorityEvaluation == TIME)
                {
                    _priority_list[i].time_priority = std::round(std::log10(std::chrono::duration_cast<std::chrono::milliseconds>(t_end - t_start).count()));
                }
                if (module_state == state::CHANGED) {
                    group_state = state::CHANGED;
                    ret = state::CHANGED;
                }
                // ignore modules which have to initilize once in the future
                if (_priority_list[i].inputCounter == 0 && _priorityEvaluation != MANUAL) {
                    restart = i;
                }
                if ((_priorityEvaluation == DYNAMIC || _priorityEvaluation == TIME) && _priority_list.size() < i+1) {
                    if (_priority_list[i + 1].inputCounter == _priority_list[i].inputCounter && _priority_list[i + 1].inputCounter != 0)
                    {
                        //collect changes
                        for (auto it = _priority_list.begin() + i + 1; it != _priority_list.end(); ++it) {
                            it->changeCounter = 0;
                            for (size_t i = 0; i < it->module_ptr->getNumInputs(); ++i) {
                                if (it->module_ptr->getInput(i)->getState() == state::CHANGED)
                                {
                                    ++(it->changeCounter);
                                }
                            }
                        }
                        std::sort(_priority_list.begin() + i + 1, _priority_list.end(), [](module_priority a, module_priority b) {
                            if (a.inputCounter == b.inputCounter)
                            {
                                if (a.time_priority == b.time_priority)
                                {
                                    if ((a.inputCounter - a.changeCounter) == (b.inputCounter - b.changeCounter)) {
                                        if (a.changeCounter == b.changeCounter) {
                                            return a.outputCounter > b.outputCounter;
                                        }
                                        return a.changeCounter > b.changeCounter;
                                    }
                                    return (a.inputCounter - a.changeCounter) < (b.inputCounter - b.changeCounter);
                                }
                                return a.time_priority < b.time_priority;
                            }
                            return a.inputCounter < b.inputCounter;
                            
                        });
                    }
                }
            }
            if (_maxIterations != -1 && processor_outputs[0]->getValue()[0] >= _maxIterations)
            {
                return state::STATE_ERROR;
            }
        }
        return ret;
    }

    std::shared_ptr<IOutput> ProcessorStandard::getIteration(){
        return processor_outputs[0];
    }
    ProcessorStandard::module_priority::module_priority(std::shared_ptr<IModule> module, size_t inputs, size_t outputs, size_t change)
    {
        module_ptr = module;
        inputCounter = inputs;
        outputCounter = outputs;
        changeCounter = change;
        time_priority = 0;
    }
}
//*/