StreamProviderI.cpp

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
 * @author
 * @date
 * @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
 *             GNU General Public License
 */
#include "StreamProviderI.h"
 
#include <cstdint>
#include <filesystem>
#include <iostream>
 
#include <ArmarXCore/core/exceptions/Exception.h>
 
#include <VisionX/interface/core/ImageProviderInterface.h>
#include <VisionX/tools/ImageUtil.h>
#include <VisionX/tools/TypeMapping.h>
 
#include <Image/ImageProcessor.h>
 
namespace Stream
{
 
    StreamProviderI::StreamProviderPropertyDefinitions::StreamProviderPropertyDefinitions(
        std::string prefix) :
        visionx::ImageProcessorPropertyDefinitions(prefix)
    {
        defineOptionalProperty<std::string>(
            "imageProviderProxyName", "TestImageProvider", "Name of the image provider");
        defineOptionalProperty<std::string>(
            "imageStreamTopicName", "ImageStream", "Name of the image streaming topic");
        defineOptionalProperty<std::string>("h264Preset", "veryfast", "Preset for the h264 codec");
        defineOptionalProperty<std::string>("h264Profile", "Main", "profile for the h264 codec");
        //            defineOptionalProperty<std::string>("OfferedSources", "Camera,Test", "Names of sources that are to be offered. For Identification on Receiver Side. seperated by comma, space or tab");
        //            defineOptionalProperty<int>("ImageWidth", 640, "Image width");
        //            defineOptionalProperty<int>("ImageHeight", 480, "Image height");
        defineOptionalProperty<float>("CompressionRate", 25, "QRC value of h264 codec");
        defineOptionalProperty<CodecType>("Codec", eH264, "Codec used for compression")
            .setCaseInsensitive(true)
            .map("h264", eH264);
        defineOptionalProperty<float>(
            "Framerate",
            25,
            "Framerate at which the input images are encoded. Slower or fasting pushing of input "
            "is possible. Frames will be skipped then. ");
    }
 
    StreamProviderI::StreamProviderI()
    {
        convertCtx = nullptr;
        encoder = nullptr;
        //    encodedImageBufferSize = 10000;
        //    encodedImageBuffer.set_capacity(encodedImageBufferSize);
        frameCounter = 0;
        srand(static_cast<unsigned int>(IceUtil::Time::now().toMicroSeconds()));
    }
 
    /// Is called once initialization of the ManagedIceObject is done.
    void
    StreamProviderI::onInitImageProcessor()
    {
        fps = getProperty<float>("Framerate");
        Logging::setTag("StreamProvider");
        offeringTopicFromProperty("imageStreamTopicName");
        // Start the timer task for the fps calcuation:
        fpsCalculator = new armarx::PeriodicTask<StreamProviderI>(
            this, &StreamProviderI::calculateFps, 995, true, "FPSCalcThread");
        fpsCalculator->start();
 
        imageProviderProxyName = getProperty<std::string>("imageProviderProxyName").getValue();
        usingImageProvider(imageProviderProxyName);
        setFramerate(fps);
        //    processTask = new armarx::PeriodicTask<StreamProviderI>(this, &StreamProviderI::processImage, 1000.0 / fps, true, "ImageEncoding", false);
 
        usingTopic("TopicRecorderListener");
    }
 
    /// Is called once all dependencies of the object have been resolved and Ice connection is established.
    void
    StreamProviderI::onConnectImageProcessor()
    {
        getTopicFromProperty(listener, "imageStreamTopicName");
 
        imageProviderProxy = getProxy<visionx::ImageProviderInterfacePrx>(imageProviderProxyName);
        visionx::ImageProviderInfo imageProviderInfo = getImageProvider(imageProviderProxyName);
 
        imageFormat = imageProviderInfo.imageFormat;
        if (imageFormat.type != visionx::eRgb)
        {
            throw armarx::LocalException("The StreamProvider supports only RGB at the moment.");
        }
        numberImages = imageProviderInfo.numberImages;
        imgWidth = imageFormat.dimension.width;
        imgHeight = imageFormat.dimension.height;
        encodedImgHeight = imgHeight * numberImages;
        imgType = imageFormat.type;
 
 
        ppInputImages = new CByteImage*[size_t(numberImages)];
        for (int i = 0; i < numberImages; i++)
        {
            ppInputImages[i] = visionx::tools::createByteImage(imageProviderInfo);
        }
 
 
        pImageForEncoder =
            new CByteImage(imageProviderInfo.imageFormat.dimension.width,
                           imageProviderInfo.imageFormat.dimension.height * numberImages,
                           visionx::tools::convert(imageProviderInfo.imageFormat.type));
 
 
        x264_param_default_preset(
            &param, getProperty<std::string>("h264Preset").getValue().c_str(), "zerolatency");
        param.i_threads = 1;
        param.i_width = imgWidth;
        param.i_height = encodedImgHeight;
        param.i_fps_num = uint32_t(fps);
        param.i_fps_den = 1;
        // Intra refres:
        param.i_keyint_max = int(fps);
        param.b_intra_refresh = 1;
        //Rate control:
        //    param.rc.i_qp_constant = 51;
        param.rc.i_rc_method = X264_RC_CRF;
        param.rc.f_rf_constant = getProperty<float>("CompressionRate");
        param.rc.f_rf_constant_max = param.rc.f_rf_constant * 1.4f;
 
        //    param.rc.i_qp_constant = 18;
        //    param.rc.i_qp_min = 18;
        //    param.rc.i_qp_max = 18;
 
        //For streaming:
        param.b_repeat_headers = 1;
        param.b_annexb = 1;
        if (x264_param_apply_profile(
                &param, getProperty<std::string>("h264Profile").getValue().c_str()) != 0)
        {
            ARMARX_WARNING << "Could not set '"
                           << getProperty<std::string>("h264Profile").getValue()
                           << "' profile for x264 codec";
        }
 
 
        startCapture();
    }
 
    /// Is called if a dependency of the object got lost (crash, network error, stopped, ...)
    void
    StreamProviderI::onDisConnectImageProcessor()
    {
        stopCapture();
 
        std::cout << "exiting StreamProviderI" << std::endl;
 
        //stop the timer task for the fps calcuation:
        fpsCalculator->stop();
 
 
        if (ppInputImages)
        {
            for (int i = 0; i < numberImages; i++)
            {
                delete ppInputImages[i];
            }
            delete[] ppInputImages;
        }
    }
 
    /// Is called once the component terminates.
    void
    StreamProviderI::onExitImageProcessor()
    {
    }
 
    void
    StreamProviderI::process()
    {
        if (!capturing)
        {
            return;
        }
        if (!waitForImages())
        {
            ARMARX_VERBOSE << "No images from provider available.";
        }
        else
        {
            armarx::MetaInfoSizeBasePtr info;
            getImages(imageProviderProxyName, ppInputImages, info);
 
 
            for (int i = 0; i < numberImages; ++i)
            {
                int imageByteSize = ppInputImages[i]->width * ppInputImages[i]->height *
                                    ppInputImages[i]->bytesPerPixel;
                memcpy(pImageForEncoder->pixels + i * imageByteSize,
                       ppInputImages[i]->pixels,
                       size_t(imageByteSize));
            }
            int srcstride = imgWidth * 3; // RGB stride is just 3*width
            const uint8_t* pixels = pImageForEncoder->pixels;
            sws_scale(convertCtx,
                      &pixels,
                      &srcstride,
                      0,
                      encodedImgHeight,
                      pic_in.img.plane,
                      pic_in.img.i_stride);
            x264_nal_t* nals;
            int i_nals;
            int frameSize = x264_encoder_encode(encoder, &nals, &i_nals, &pic_in, &pic_out);
            if (frameSize >= 0)
            {
                ARMARX_DEBUG << deactivateSpam(1) << "encoded image: " << VAROUT(frameSize);
 
                frameCounter++;
 
                DataChunk chunk;
                chunk.reserve(size_t(frameSize));
                chunk.assign(nals->p_payload, nals->p_payload + frameSize);
                listener->reportNewStreamData(chunk, info->timeProvided);
            }
        }
    }
 
    /**
     * Retrieve default name of component
     *
     * @return default name of the component
     */
    std::string
    StreamProviderI::getDefaultName() const
    {
        return "StreamProvider";
    }
 
    bool
    StreamProviderI::startCapture(const ::Ice::Current&)
    {
 
        if (capturing)
        {
            stopCapture();
        }
        encoder = x264_encoder_open(&param);
 
        x264_picture_alloc(&pic_in, X264_CSP_I420, imgWidth, encodedImgHeight);
 
        convertCtx = sws_getContext(imgWidth,
                                    encodedImgHeight,
                                    AV_PIX_FMT_RGB24,
                                    imgWidth,
                                    encodedImgHeight,
                                    AV_PIX_FMT_YUV420P,
                                    SWS_FAST_BILINEAR,
                                    nullptr,
                                    nullptr,
                                    nullptr);
 
        capturing = true;
 
        return true;
    }
 
    void
    StreamProviderI::stopCapture(const ::Ice::Current&)
    {
        capturing = false;
 
        if (encoder)
        {
            x264_picture_clean(&pic_in);
            memset(reinterpret_cast<char*>(&pic_in), 0, sizeof(pic_in));
            memset(reinterpret_cast<char*>(&pic_out), 0, sizeof(pic_out));
 
            x264_encoder_close(encoder);
            encoder = nullptr;
        }
 
        if (convertCtx)
        {
            sws_freeContext(convertCtx);
            convertCtx = nullptr;
        }
    }
 
    CodecType
    StreamProviderI::getCodecType(const Ice::Current&)
    {
        return getProperty<CodecType>("Codec").getValue();
    }
 
    void
    StreamProviderI::calculateFps()
    {
        //    StreamSourceMap::iterator it = streamSources.begin();
        //    for(; it != streamSources.end(); it++)
        //    {
        //        it->second->fps = it->second->fetchedChunks;//0.2*fetchedChunks+0.8*fps; //calculate the new fps as weighted medium
        //        it->second->fps = it->second->fps > 10 ? (it->second->fps+2) : 10; //set the fps to a minimum of 10, increase the fps, so that slightly more images can be pushed in than pulled out
        //        armarx::ScopedLock lock(it->second->mutex);
        //        it->second->fetchedChunks = 0;
        //    }
    }
 
    void
    StreamProviderI::getImageInformation(int& imageWidth,
                                         int& imageHeight,
                                         int& imageType,
                                         const Ice::Current&)
    {
        imageWidth = imgWidth;
        imageHeight = imgHeight;
        imageType = imgType;
    }
 
    int
    StreamProviderI::getNumberOfImages(const Ice::Current&)
    {
        return numberImages;
    }
 
    void
    StreamProviderI::setCompressionRate(CompressionRate, const Ice::Current&)
    {
        ARMARX_WARNING << "Not yet implemented";
    }
 
    void
    StreamProviderI::onStartRecording(const Ice::Current&)
    {
        startCapture();
    }
 
} // namespace Stream