StreamReceiver.h

Go to the documentation of this file.

/**
 * This file is part of ArmarX.
 *
 * Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
 *
 * 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    TabletTeleoperation::ArmarXObjects::StreamReceiver
 * @author     Mirko Waechter ( mirko dot waechter at kit dot edu )
 * @date       2013
 * @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
 *             GNU General Public License
 */
 
#pragma once
 
 
#include <ArmarXCore/core/Component.h>
#include <ArmarXCore/core/services/tasks/PeriodicTask.h>
#include <VisionX/interface/components/StreamProvider.h>
#include <VisionX/components/image_processor/StreamProvider/StreamProviderI.h>
 
extern "C"
{
    class _GstElement;
    using GstElement = _GstElement;
}
 
extern "C" {
#include <libavcodec/avcodec.h>
#include <libswscale/swscale.h>
}
 
namespace armarx
{
 
 
    using ImageMap = std::map<std::string, CByteImage*>;
    /**
         * @class StreamReceiver
         * @brief A brief description
         *
         * Detailed Description
         */
    class StreamReceiver :
        virtual public armarx::Component,
        public Stream::StreamListenerInterface
    {
    public:
        class StreamReceiverPropertyDefinitions:
            public armarx::ComponentPropertyDefinitions
        {
        public:
            StreamReceiverPropertyDefinitions(std::string prefix):
                ComponentPropertyDefinitions(prefix)
            {
                defineOptionalProperty<std::string>("UsedStreamProvider", "StreamProvider", "Names of the StreamProvider that is to be used");
                //  defineOptionalProperty<std::string>("UsedSources", "Camera,Test", "Names of sources that are to be used of the StreamProvider. Seperated by comma, space or tab");
            }
        };
 
        /**
             * @see armarx::ManagedIceObject::getDefaultName()
             */
        virtual std::string getDefaultName() const
        {
            return "StreamReceiver";
        }
 
        /**
             * @brief getImages retreives the next buffered images for all streams.
             * @param imagesOut map of already allocated images in correct size
             */
        void getImages(std::vector<CByteImage*>& imagesOut);
 
        /**
         * @brief getNumberOfImages can be called after the component was initialized.
         * @return the number of provided images
         */
        int getNumberOfImages();
 
 
        void getImageInformation(int& imageWidth, int& imageHeight, int& imageType);
 
 
    protected:
        /**
             * @see armarx::ManagedIceObject::onInitComponent()
             */
        virtual void onInitComponent();
 
        /**
             * @see armarx::ManagedIceObject::onConnectComponent()
             */
        virtual void onConnectComponent();
 
        /**
             * @see armarx::ManagedIceObject::onDisconnectComponent()
             */
        virtual void onDisconnectComponent();
 
        /**
             * @see armarx::ManagedIceObject::onExitComponent()
             */
        virtual void onExitComponent();
 
 
        /**
         * @see PropertyUser::createPropertyDefinitions()
         */
        virtual armarx::PropertyDefinitionsPtr createPropertyDefinitions()
        {
            return armarx::PropertyDefinitionsPtr(
                       new StreamReceiverPropertyDefinitions(
                           getConfigIdentifier()));
        }
    private:
 
        void receive();
        void store();
 
        class StreamElements;
        using StreamElementsPtr = std::shared_ptr<StreamElements>;
        struct StreamElements  : Stream::StreamMetaData
        {
        public:
            StreamElements(std::string streamName, int streamID, Stream::StreamProviderPrx streamProvider);
            GstElement* pipeline;
            GstElement* appsrc;
            GstElement* appsink;
            std::stringstream pipelineString;
            armarx::Mutex mutex;
            int fetchedChunks;
            IceUtil::Time lastFPSCheck;
            int framesSinceLastFPSCheck;
            PeriodicTask<StreamElements>::pointer_type taskReceive;
            Stream::StreamProviderPrx streamProvider;
            void receive();
            int receivedIndex;
            int realPosition;
            int streamID;
 
            boost::mutex mutexPulling;
 
        };
        using StreamSourceMap = boost::unordered_map<std::string, StreamElementsPtr>;
        StreamReceiver::StreamElementsPtr getStreamElements(std::string streamName);
        void getImageFormat(StreamElementsPtr elem, int& height, int& width);
 
 
        Stream::StreamProviderPrx streamProvider;
        //    PeriodicTask<StreamReceiver>::pointer_type taskReceive;
        PeriodicTask<StreamReceiver>::pointer_type taskSave;
        StreamSourceMap streamSources;
 
 
        int numberImages;
 
        IceUtil::Time start;
        long transferredBits;
        //    GstElement* appsrc;
        //    GstElement* appsink;
        //    GstElement* pipeline;
        float bandwidth_kbps;
        Mutex pipelineMutex;
        Stream::CodecType codec;
 
        IceUtil::Time lastReceiveTimestamp;
 
 
        // StreamListenerInterface interface
    public:
        void reportNewStreamData(const Stream::DataChunk& chunk, const Ice::Current&);
 
        AVCodec* m_decoder;
        AVCodecContext* m_decoderContext;
        int m_got_picture;
        AVFrame* m_picture;
        AVPacket m_packet;
        Mutex decodedImageMutex;
        Mutex streamDecodeMutex;
        CByteImage** ppDecodedImages, *pCombinedDecodedImage;
    };
    using StreamReceiverPtr = IceInternal::Handle<StreamReceiver>;
}