Collector.h

Go to the documentation of this file.

/* ============================================================
 *
 * This file is a part of CCA project
 *
 * Copyright (C) 2011 by Arne Nordmann <anordman at cor-lab dot uni-bielefeld dot de>
 *
 * This file may be licensed under the terms of the
 * GNU Lesser General Public License Version 3 (the ``LGPL''),
 * or (at your option) any later version.
 *
 * Software distributed under the License is distributed
 * on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
 * express or implied. See the LGPL for the specific language
 * governing rights and limitations.
 *
 * You should have received a copy of the LGPL along with this
 * program. If not, go to http://www.gnu.org/licenses/lgpl.html
 * or write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 * The development of this software was supported by:
 *   CoR-Lab, Research Institute for Cognition and Robotics
 *     Bielefeld University
 *
 * ============================================================ */

#pragma once

#include <vector>

#include <string>
#include <iostream>
#include <sstream>

#include <rsc/logging/Logger.h>

#include <nemo/Vector.h>

#include "cca/Node.h"
#include "cca/processing/Moderate.h"

namespace cca {

template<class DTOTYPE>
class Collector: public Node {
public:

    Collector(std::string nodename, unsigned int number) :
            Node(nodename), dim(number),
                    collectConfig(nemo::IntVector(nemo::dim(number), 1)), ips(),
                    op(),
                    logger(rsc::logging::Logger::getLogger("cca.collector")) {

        for (unsigned int port = 0; port < dim; ++port) {
            ips.push_back(InputPort<DTOTYPE>::create());
            std::stringstream oss;
            oss << "in" << port;
            registerPort(oss.str(), ips[port]);
        }

        op = OutputPort<DTOTYPE>::create();
        registerPort("out", op);

        // Define default strategy - moderate
        this->setProcessingStrategy(Moderate::timeout());
    }

    Collector(std::string nodename, nemo::IntVector config) :
            Node(nodename), dim(config.dimension()), collectConfig(config),
                    ips(), op(),
                    logger(rsc::logging::Logger::getLogger("cca.collector")) {

        unsigned int count = 0;
        for (unsigned int port = 0; port < collectConfig.dimension(); ++port) {
            ips.push_back(InputPort<DTOTYPE>::create());
            std::stringstream oss;
            oss << "in" << port;
            registerPort(oss.str(), ips[port]);
            count += collectConfig[port];
        }
        this->dim = count;

        op = OutputPort<DTOTYPE>::create();
        registerPort("out", op);

        // Define default strategy - moderate
        this->setProcessingStrategy(Moderate::timeout());
    }

    ~Collector() {
    }

    void onProcess() {
        std::vector<boost::shared_ptr<DTOTYPE> > collected = std::vector<
                boost::shared_ptr<DTOTYPE> >();

        // Collect values from input - none of the inputs can be empty
        for (unsigned int port = 0; port < collectConfig.dimension(); ++port) {
            if (!ips[port]->empty()) {
                collected.push_back(ips[port]->get());
            } else {
                logger->error(
                        "Collector " + this->getName()
                                + "onProcess() Not all ports received data yet, can`t collect and merge.");
                return;
            }
        }

        // Preparing and sending collected DTO
        typename OutputPort<DTOTYPE>::DataPtr merged = DTOTYPE::create(
                this->dim);
        merged->merge(collected);
        op->publish(merged);
    }

    static NodePtr create(std::string nodename, unsigned int number) {
        return NodePtr(new Collector<DTOTYPE>(nodename, number));
    }

    static NodePtr create(std::string nodename, nemo::IntVector config) {
        return NodePtr(new Collector<DTOTYPE>(nodename, config));
    }

    static NodePtr create(std::string nodename, int first, int second) {
        return NodePtr(
                new Collector<DTOTYPE>(nodename,
                        nemo::IntVector(first, second)));
    }

    static NodePtr create(std::string nodename, int first, int second,
            int third) {
        return NodePtr(
                new Collector<DTOTYPE>(nodename,
                        nemo::IntVector(first, second, third)));
    }

    static NodePtr create(unsigned int number) {
        return NodePtr(new Collector<DTOTYPE>("Collector", number));
    }

    friend std::ostream & operator<<(std::ostream& os,
            const Collector<DTOTYPE> &node) {
        os.precision(3); // Precision when printing double values
        os << std::endl << "<Collector '" << node.getName() << "'>"
                << std::endl;
        return os;
    }

    virtual void configureInputPortByIndex(unsigned int index,
            PortConfigurationPtr portcfg) {
        std::stringstream portname;
        portname << "in" << index;
        this->configureInputPort(portname.str(), portcfg);
    }

    virtual void configureInputPortByIndex(unsigned int index,
            const std::string &scope) {
        std::stringstream portname;
        portname << "in" << index;
        this->configureInputPort(portname.str(), scope);
    }

private:
    unsigned int dim;
    nemo::IntVector collectConfig;
    std::vector<typename InputPort<DTOTYPE>::Ptr> ips;
    typename OutputPort<DTOTYPE>::Ptr op;
    rsc::logging::LoggerPtr logger;
};

}