CamManager.cpp

Go to the documentation of this file.

#include <CamManager.h>

#include <ZmqPubExtractor.h>
#include <FileStorageExtractor.h>
#include <StatsExtractor.h>
#include <FFMpegExtractor.h>
#include <stringtricks.h>
#include <config.h>

#include <iostream>

using namespace std;

CamManager::CamManager(zmq::context_t & _ctx,
                       const std::string & _brokerEndpoint,
                       const std::string & _serviceName,
                       const PubSubEndpoint & _cameraSource,
                       StorageManager & _storageManager):
    RocServer(_ctx, _brokerEndpoint, _serviceName),
    cameraSource(_cameraSource),
    logger(ctx, "inproc://log", "camserver", SSTR("camManager.", cameraSource.topic)) {
    storageManager = &_storageManager;
}

CamManager::~CamManager() {
    unique_lock<mutex> _(extractorsMutex);
    for(auto extractorP: extractors) {
        extractorP->stopServing();
        delete extractorP;
    }
}

Json::Value CamManager::call(const std::string & name, const Json::Value & arguments) {
    if(name == "addExtractor") {
        return addExtractor(arguments[0].asString(), arguments[1]);
    } else if(name == "listExtractors") {
        return listExtractors();
    } else if(name == "removeExtractor") {
        return removeExtractor(arguments[0].asString(), arguments[1].asString());
    }
    throw RocNotImplementedError();
}


Json::Value CamManager::addExtractor(const std::string & name, const Json::Value & arguments) {
    logger.info(SSTR("add Extractor: ", name, " (", arguments, ") "));
    auto identifier = arguments[0].asString();
    {
        unique_lock<mutex> _(extractorsMutex);
        for(auto extractorP: extractors) {
            if(extractorP->type() == name && extractorP->identifier() == identifier) {
                /* already registered! */
                logger.debug("already registered!");
                return "already registered!";
            }
        }
    }
    Extractor * extractor;
    if(name == "zmqPub") {
        extractor = new ZmqPubExtractor(ctx,
                                        cameraSource,
                                        {"inproc://output", identifier},
                                        {arguments[1][0].asUInt(),
                                         arguments[1][1].asUInt(),
                                         arguments[1][2].asUInt(),
                                         arguments[1][3].asUInt()});
    } else if(name == "fileStorage") {
        int maxFrames = -1; /* infinite */
        if(arguments.size() > 1) {
            maxFrames = arguments[1].asInt();
        }
        string metaData = "{}";
        if(arguments.size() > 2) {
            metaData = arguments[2].asString();
        }
        int fileCycling = -1;
        if(arguments.size() > 3) {
            fileCycling = arguments[3].asInt();
        }
        bool cumulative = false;
        if(arguments.size() > 4) {
            cumulative = arguments[4].asBool();
        }
        bool binning = false;
        if(arguments.size() > 5) {
            if(cameraSource.topic.compare(0, 4, "vnir") == 0)
                binning = arguments[5].asBool();
        }
        double wantedFPS = 0;
        if(arguments.size() > 6) {
            wantedFPS = arguments[6].asDouble();
        }
        extractor = new FileStorageExtractor(ctx, cameraSource, identifier, *storageManager, maxFrames, metaData, fileCycling, cumulative, binning, wantedFPS);
    } else if(name == "Stats") {
        extractor = new StatsExtractor(ctx,
                                       cameraSource,
                                       {"inproc://log", identifier},
                                       identifier);
    } else if(name == "FFMpeg") {
        auto encode_to = arguments[1];
        extractor = new FFMpegExtractor(ctx,
                                        cameraSource,
                                        identifier,
                                        encode_to.asString());

    } else {
        throw RocNotImplementedError();
    }
    extractor->startServing();
    unique_lock<mutex> _(extractorsMutex);
    extractors.push_back(extractor);
    return Json::Value::null;
}

Json::Value CamManager::listExtractors() {
    /* cout << "list Extractors" << endl; */
    Json::Value ret(Json::ValueType::arrayValue);
    unique_lock<mutex> _(extractorsMutex);
    for(auto extractorP: extractors) {
        Json::Value exId;
        exId["type"] = extractorP->type();
        exId["identifier"] = extractorP->identifier();
        Json::Value params;
        for(auto param: extractorP->parameters()) {
            params.append(param);
        }
        exId["parameters"] = params;
        ret.append(exId);
    }
    return ret;
}


Json::Value CamManager::removeExtractor(const std::string & name, const std::string & identifier) {
    bool removedSomething = false;
    logger.info(SSTR("removing ", name, " / ", identifier, " extractor."));
    {
        unique_lock<mutex> _(extractorsMutex);
        for(auto it = extractors.begin(); it != extractors.end();) {
            if((*it)->type() == name && (*it)->identifier() == identifier) {
                (*it)->stopServing();
                delete (*it);
                removedSomething = true;
                it = extractors.erase(it);
            } else {
                ++it;
            }
        }
    }
    cout << "done removing." << endl;
    if(removedSomething) {
        return Json::Value::null;
    } else {
        return "not found";
    }
}