<commit_before>/// PROJECT
#include <csapex/model/node.h>
#include <csapex/utility/register_apex_plugin.h>
#include <csapex/model/node_modifier.h>
#include <csapex/param/parameter_factory.h>
#include <csapex/model/variadic_io.h>
#include <csapex/msg/io.h>
#include <csapex/msg/generic_vector_message.hpp>
#include <csapex/signal/slot.h>
#include <csapex/msg/input.h>
#include <csapex/msg/message.h>
#include <csapex_ros/ros_message_conversion.h>
#include <csapex/model/node_handle.h>

/// SYSTEM
#include <rosbag/bag.h>

namespace csapex
{

class CSAPEX_EXPORT_PLUGIN BagWriter : public Node, public VariadicInputs, public VariadicSlots
{
public:
    BagWriter()
        : is_open_(false)
    {}

    ~BagWriter()
    {
        stop();
    }

    void setup(csapex::NodeModifier& node_modifier)
    {
        setupVariadic(node_modifier);
    }

    void setupVariadicParameters(Parameterizable &parameters)
    {
        VariadicInputs::setupVariadicParameters(parameters);
        VariadicSlots::setupVariadicParameters(parameters);
    }

    void setupParameters(Parameterizable& parameters)
    {
        setupVariadicParameters(parameters);

        parameters.addParameter(param::ParameterFactory::declareFileOutputPath("file", "/tmp/bag.bag", "*.bag"), [this](param::Parameter* p) {
            bag.open(p->as<std::string>(), rosbag::bagmode::Write);
            is_open_ = true;
        });

        parameters.addParameter(param::ParameterFactory::declareTrigger("reset"), [this](param::Parameter* p) {
            stop();
            bag.open(readParameter<std::string>("file"), rosbag::bagmode::Write);
            is_open_ = true;
        });

        parameters.addParameter(param::ParameterFactory::declareTrigger("save and close"), [this](param::Parameter* p) {
            stop();
        });
    }
    void process()
    {
        if(!is_open_) {
            node_handle_->setWarning("bag file is closed, cannot write");
            return;
        } else {
            node_handle_->setNoError();
        }

        std::vector<InputPtr> inputs = node_modifier_->getMessageInputs();
        for(std::size_t i = 0 ; i < inputs.size() ; i++) {
            Input *in = inputs[i].get();
            if(msg::hasMessage(in)) {
                connection_types::MessageConstPtr message = msg::getMessage<connection_types::Message>(in);

                RosMessageConversion::instance().write(bag, message, in->getLabel());
            }
        }
    }

    void stop()
    {
        bag.close();
        is_open_ = false;
    }

    virtual csapex::Input* createVariadicInput(csapex::TokenDataConstPtr type, const std::string& label, bool optional) override
    {
        return VariadicInputs::createVariadicInput(connection_types::makeEmpty<connection_types::AnyMessage>(), label.empty() ? "Value" : label, getVariadicInputCount() == 0 ? false : true);
    }

    virtual csapex::Slot* createVariadicSlot(TokenDataConstPtr type, const std::string& label, std::function<void (const TokenPtr&)> callback, bool active, bool asynchronous) override
    {
        auto cb = [this](Slot* slot, const TokenPtr& token) {
            auto message = std::dynamic_pointer_cast<connection_types::Message const>(token->getTokenData());
            RosMessageConversion::instance().write(bag, message, slot->getLabel());
        };

        return VariadicSlots::createVariadicSlot(connection_types::makeEmpty<connection_types::AnyMessage>(), label.empty() ? "Value" : label, cb, active, asynchronous);
    }

    void write(const std::string& label, long stamp, std::shared_ptr<topic_tools::ShapeShifter const> instance)
    {
        if(instance) {
            apex_assert(!instance->getMD5Sum().empty());
            if(stamp > 0) {
                ros::Time time;
                time.fromNSec(stamp * 1e3);
                bag.write(label, time, *instance);
            }
        } else {
            aerr << "cannot write " << label << ", instance is null" << std::endl;
        }
    }


    virtual Connectable* createVariadicPort(ConnectorType port_type, TokenDataConstPtr type, const std::string& label, bool optional) override
    {
        apex_assert_hard(variadic_modifier_);
        switch (port_type) {
        case ConnectorType::INPUT:
            return createVariadicInput(type, label, optional);
        case ConnectorType::SLOT_T:
            return createVariadicSlot(type, label, [](const TokenPtr&){}, false, false);
        default:
            throw std::logic_error(std::string("Variadic port of type ") + port_type::name(port_type) + " is not supported.");
        }
    }

private:
    Output* output;

    rosbag::Bag bag;
    bool is_open_;
};

}

CSAPEX_REGISTER_CLASS(csapex::BagWriter, csapex::Node)<commit_msg>WriteBag: new file if new name is provided<commit_after>/// PROJECT
#include <csapex/model/node.h>
#include <csapex/utility/register_apex_plugin.h>
#include <csapex/model/node_modifier.h>
#include <csapex/param/parameter_factory.h>
#include <csapex/model/variadic_io.h>
#include <csapex/msg/io.h>
#include <csapex/msg/generic_vector_message.hpp>
#include <csapex/signal/slot.h>
#include <csapex/msg/input.h>
#include <csapex/msg/message.h>
#include <csapex_ros/ros_message_conversion.h>
#include <csapex/model/node_handle.h>

/// SYSTEM
#include <rosbag/bag.h>

namespace csapex
{

class CSAPEX_EXPORT_PLUGIN BagWriter : public Node, public VariadicInputs, public VariadicSlots
{
public:
    BagWriter()
        : is_open_(false)
    {}

    ~BagWriter()
    {
        stop();
    }

    void setup(csapex::NodeModifier& node_modifier)
    {
        setupVariadic(node_modifier);
    }

    void setupVariadicParameters(Parameterizable &parameters)
    {
        VariadicInputs::setupVariadicParameters(parameters);
        VariadicSlots::setupVariadicParameters(parameters);
    }

    void setupParameters(Parameterizable& parameters)
    {
        setupVariadicParameters(parameters);

        parameters.addParameter(param::ParameterFactory::declareFileOutputPath("file", "/tmp/bag.bag", "*.bag"), [this](param::Parameter* p) {
            std::string file_name = p->as<std::string>();
            if(file_name == file_name_){
                return;
            }
            file_name_ = file_name;
            start();
        });

        parameters.addParameter(param::ParameterFactory::declareTrigger("reset"), [this](param::Parameter* p) {
            start();

        });

        parameters.addParameter(param::ParameterFactory::declareTrigger("save and close"), [this](param::Parameter* p) {
            stop();
        });
    }
    void process()
    {
        if(!is_open_) {
            node_handle_->setWarning("bag file is closed, cannot write");
            return;
        } else {
            node_handle_->setNoError();
        }

        std::vector<InputPtr> inputs = node_modifier_->getMessageInputs();
        for(std::size_t i = 0 ; i < inputs.size() ; i++) {
            Input *in = inputs[i].get();
            if(msg::hasMessage(in)) {
                connection_types::MessageConstPtr message = msg::getMessage<connection_types::Message>(in);

                RosMessageConversion::instance().write(bag, message, in->getLabel());
            }
        }
    }

    void stop()
    {
        if(is_open_){
            bag.close();
            is_open_ = false;
        }
    }

    void start()
    {
        try{
            if(is_open_){
                stop();
            }
            bag.open(file_name_, rosbag::bagmode::Write);
            is_open_ = true;
        }
        catch(const rosbag::BagException& e){
            is_open_ = false;
            node_handle_->setError(e.what());
        }
    }

    virtual csapex::Input* createVariadicInput(csapex::TokenDataConstPtr type, const std::string& label, bool optional) override
    {
        return VariadicInputs::createVariadicInput(connection_types::makeEmpty<connection_types::AnyMessage>(), label.empty() ? "Value" : label, getVariadicInputCount() == 0 ? false : true);
    }

    virtual csapex::Slot* createVariadicSlot(TokenDataConstPtr type, const std::string& label, std::function<void (const TokenPtr&)> callback, bool active, bool asynchronous) override
    {
        auto cb = [this](Slot* slot, const TokenPtr& token) {
            auto message = std::dynamic_pointer_cast<connection_types::Message const>(token->getTokenData());
            RosMessageConversion::instance().write(bag, message, slot->getLabel());
        };

        return VariadicSlots::createVariadicSlot(connection_types::makeEmpty<connection_types::AnyMessage>(), label.empty() ? "Value" : label, cb, active, asynchronous);
    }

    void write(const std::string& label, long stamp, std::shared_ptr<topic_tools::ShapeShifter const> instance)
    {
        if(instance) {
            apex_assert(!instance->getMD5Sum().empty());
            if(stamp > 0) {
                ros::Time time;
                time.fromNSec(stamp * 1e3);
                bag.write(label, time, *instance);
            }
        } else {
            aerr << "cannot write " << label << ", instance is null" << std::endl;
        }
    }


    virtual Connectable* createVariadicPort(ConnectorType port_type, TokenDataConstPtr type, const std::string& label, bool optional) override
    {
        apex_assert_hard(variadic_modifier_);
        switch (port_type) {
        case ConnectorType::INPUT:
            return createVariadicInput(type, label, optional);
        case ConnectorType::SLOT_T:
            return createVariadicSlot(type, label, [](const TokenPtr&){}, false, false);
        default:
            throw std::logic_error(std::string("Variadic port of type ") + port_type::name(port_type) + " is not supported.");
        }
    }

private:
    Output* output;

    rosbag::Bag bag;
    bool is_open_;
    std::string file_name_;
};

}

CSAPEX_REGISTER_CLASS(csapex::BagWriter, csapex::Node)
{'fraction_non_alphanumeric': 0.13204882477554727, 'fraction_numerical': 0.0012105316251387068, 'mean_word_length': 2.878716744913928, 'pattern_counts': {'":': 0, '<': 64, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3840539', 'n_tokens_mistral': 3044, 'n_tokens_neox': 2921, 'n_words': 533}