TopicReplayer.cpp

Go to the documentation of this file.

/*
* This file is part of ArmarX.
*
* ArmarX is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation.
*
* ArmarX is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* @package    ArmarX
* @author     Mirko Waechter( mirko.waechter at kit dot edu)
* @date       2016
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
 
#include "TopicReplayer.h"
 
#include <string>
 
#include <IceStorm/IceStorm.h>
 
#include <ArmarXCore/core/ArmarXManager.h>
#include <ArmarXCore/core/time/TimeUtil.h>
#include <ArmarXCore/util/json/JSONObject.h>
 
#include "DatabaseTopicReader.h"
#include "FileTopicReader.h"
#include "TopicUtil.h"
 
namespace armarx
{
    TopicReplayer::TopicReplayer()
    {
    }
 
    PropertyDefinitionsPtr
    TopicReplayer::createPropertyDefinitions()
    {
        return new TopicReplayerProperties(Component::getConfigIdentifier());
    }
 
    void
    TopicReplayer::onInitComponent()
    {
        loop = getProperty<bool>("Loop").getValue();
        //Check which storage mode to use and init the reader
        std::string storageMode = getProperty<std::string>("StorageMode").getValue();
        ARMARX_INFO << "reading from file: " << getProperty<std::string>("RecordFile").getValue();
        if (!storageMode.compare("file"))
        {
            replayer.reset(new FileTopicReader(getProperty<std::string>("RecordFile").getValue()));
        }
        else if (!storageMode.compare("database"))
        {
            replayer.reset(
                new DatabaseTopicReader(getProperty<std::string>("RecordFile").getValue()));
        }
        else
        {
            ARMARX_WARNING
                << "StorageMode " << storageMode
                << " is not supported (database, file). Falling back to default 'database' mode.";
            replayer.reset(
                new DatabaseTopicReader(getProperty<std::string>("RecordFile").getValue()));
        }
 
        offeringTopic(getProperty<std::string>("TopicName"));
        offeringTopic(getProperty<std::string>("DebugObserverName").getValue());
    }
 
    void
    TopicReplayer::onConnectComponent()
    {
        debugObserver = getTopic<DebugObserverInterfacePrx>(
            getProperty<std::string>("DebugObserverName").getValue());
 
        timeKeeper.reset();
        timeKeeper.setSpeed(1.0);
 
        replayerListener =
            getTopic<TopicReplayerListenerInterfacePrx>(getProperty<std::string>("TopicName"));
 
        task = new RunningTask<TopicReplayer>(this, &TopicReplayer::play, "ReplayThread");
        task->start();
        if (autoplay)
        {
            setReplayingTopics(replayer->getReplayTopics());
            resumeReplay();
        }
    }
 
    std::string
    TopicReplayer::getDefaultName() const
    {
        return "TopicReplayer";
    }
 
    void
    TopicReplayer::play()
    {
        std::map<std::string, Ice::ObjectPrx> topics;
        TopicUtil::TopicData data;
 
        while (!task->isStopped())
        {
            if (replayer->read(data))
            {
                std::string topic = data.topicName;
                if (topics.count(topic) == 0)
                {
                    Ice::ObjectPrx topicPrx = getTopic<Ice::ObjectPrx>(topic);
                    topics[topic] = topicPrx;
                }
                Ice::ObjectPrx t = topics[topic];
                std::vector<Ice::Byte> dataBytesOut;
 
                while (this->timeKeeper.getTime() < data.timestamp && !task->isStopped())
                {
                    usleep(100);
                }
                if (this->replayingTopicsNotSupportedByFile ||
                    replayingTopics.count(data.topicName))
                {
                    t->ice_invoke(data.operationName, Ice::Normal, data.inParams, dataBytesOut);
                }
            }
            else if (loop)
            {
                if (!replayer->seekTo(IceUtil::Time::seconds(0.0)))
                {
                    ARMARX_WARNING << "unable to rewind ";
                    task->stop();
                    timeKeeper.stop();
                }
                else
                {
                    ARMARX_INFO << "rewinding...";
                    timeKeeper.reset();
 
                    StringVariantBaseMap debugValues;
                    debugValues["file"] =
                        new Variant(getProperty<std::string>("RecordFile").getValue());
                    debugValues["status"] = new Variant("rewinding");
                    debugObserver->setDebugChannel(getName(), debugValues);
                }
            }
            else
            {
                task->stop();
                timeKeeper.stop();
                replayerListener->onStopReply();
 
                StringVariantBaseMap debugValues;
                debugValues["status"] = new Variant("stopped");
                debugObserver->setDebugChannel(getName(), debugValues);
 
                if (autoplay)
                {
                    getArmarXManager()->asyncShutdown();
                }
            }
 
            // StringVariantBaseMap debugValues;
            // debugValues["replay_time"] = new Variant(timeKeeper.getTime());
            // debugObserver->setDebugChannel(getName(), debugValues);
        }
    }
 
    void
    TopicReplayer::onDisconnectComponent()
    {
        if (task)
        {
            task->stop();
        }
    }
 
    void
    TopicReplayer::pauseReplay()
    {
        this->timeKeeper.stop();
 
        StringVariantBaseMap debugValues;
        debugValues["status"] = new Variant("paused");
        debugObserver->setDebugChannel(getName(), debugValues);
    }
 
    void
    TopicReplayer::resumeReplay()
    {
        if (task->isStopped())
        {
            task->start();
        }
        this->timeKeeper.start();
        this->replayerListener->onStartReplay(getProperty<std::string>("RecordFile").getValue());
 
        StringVariantBaseMap debugValues;
        debugValues["file"] = new Variant(getProperty<std::string>("RecordFile").getValue());
        debugValues["status"] = new Variant("started");
        debugObserver->setDebugChannel(getName(), debugValues);
    }
 
    void
    TopicReplayer::setReplaySpeed(double factor)
    {
        this->timeKeeper.setSpeed((float)factor);
    }
 
    void
    TopicReplayer::jumpToPosition(IceUtil::Time timestamp)
    {
        ARMARX_INFO << "Jump called with TimeStamp: " << timestamp.toDuration();
        this->timeKeeper.step(timestamp - this->timeKeeper.getTime());
        replayer->seekTo(timestamp);
 
        task->stop();
        task = new RunningTask<TopicReplayer>(this, &TopicReplayer::play, "ReplayThread");
        task->start();
    }
 
    std::vector<std::string>
    TopicReplayer::getRecordedTopics()
    {
        std::vector<std::string> topics = replayer->getReplayTopics();
        if (topics.empty())
        {
            replayingTopicsNotSupportedByFile = true;
        }
 
        return topics;
    }
 
    void
    TopicReplayer::setReplayingTopics(std::vector<std::string> topics)
    {
        this->replayingTopics = std::unordered_set<std::string>(topics.begin(), topics.end());
    }
 
    IceUtil::Time
    TopicReplayer::getReplayLength() const
    {
        return replayer->getReplayLength();
    }
 
    IceUtil::Time
    TopicReplayer::getCurrentTimePosition() const
    {
        return timeKeeper.getTime();
    }
 
    void
    TopicReplayer::setAutoplay(bool autoplay)
    {
        this->autoplay = autoplay;
    }
 
    void
    TopicReplayer::setLoop(bool loop)
    {
        this->loop = loop;
    }
} // namespace armarx