camserver.cpp

Go to the documentation of this file.

#include <iostream>
#include <atomic>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <chrono>
#include <vector>
#include <zmq.hpp>

#include <tclap/CmdLine.h>
#include <uuid.h>

#include <ImageGrabber.h>
#include <ZmqPubExtractor.h>
#include <StatsExtractor.h>
#include <FFMpegExtractor.h>
#include <RocLogger.h>
#include <stringtricks.h>
#include <CamManager.h>
#include <InfoServer.h>
#include <NetworkHelper.h>
#include <ConfigManager.h>
#include <StorageManager.h>

#include <json/json.h>

#include <config.h>

#ifdef HAVE_WINDOWS
#include <windows.h>
#endif
#ifdef HAVE_LINUX
#include <signal.h>
#include <unistd.h>
#include <cstdlib>
#endif
#include <cstdio>

using namespace std;

static volatile bool keepRunning;

#ifdef HAVE_WINDOWS
BOOL CtrlHandler(DWORD fdwCtrlType) {
    switch(fdwCtrlType) {
        case CTRL_C_EVENT:
            if(keepRunning) {
                cout << "stopping server, please wait ..." << endl;
                keepRunning = false;
                return TRUE;
            } else {
                cout << "received ctrl+c twice: terminating!" << endl;
                return FALSE;
            }
        case CTRL_BREAK_EVENT:
            if(keepRunning) {
                cout << "stopping server, please wait ..." << endl;
                keepRunning = false;
                return TRUE;
            } else {
                cout << "received ctrl+c twice: terminating!" << endl;
                return FALSE;
            }
        default:
            return FALSE;
    }
}
#endif

#ifdef HAVE_LINUX
void SignalHandler(int s) {
    switch(s) {
        case SIGINT:
            if(keepRunning) {
                cout << "stopping server, please wait ..." << endl;
                keepRunning = false;
                return;
            } else {
                cout << "received ctrl+c twice: terminating!" << endl;
                abort();
                return;
            }
        default:
            return;
    }
}
#endif

void pubsubproxy(zmq::context_t * ctx,
                 const string & from,
                 const string & to,
                 bool connectTo = false,
                 condition_variable * isUp = nullptr,
                 int linger = 100) {
                 /* a positive linger value ensures that the proxy will not hang forever on network errors */
    zmq::socket_t in(*ctx, ZMQ_XSUB);
    zmq::socket_t out(*ctx, ZMQ_XPUB);
    in.bind(from.c_str());
    out.setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
    if(connectTo) {
        out.connect(to.c_str());
    } else {
        out.bind(to.c_str());
    }
    if(isUp != nullptr) {
        isUp->notify_all();
    }
    try {
        zmq::proxy((void*)in, (void*)out, NULL);
    } catch(zmq::error_t &e) {
        cout << e.what() << endl;
    }
    cout << "closing proxy" << endl;
}

void sendControlMessage(zmq::socket_t & socket, const string & msg) {
    zmq::message_t topic(7);
    memcpy(topic.data(), "control", 7);
    socket.send(topic, ZMQ_SNDMORE);
    zmq::message_t data(msg.length());
    memcpy(data.data(), msg.c_str(), msg.length());
    socket.send(data);
}

int main(int argc, char** argv) {
    try {
        TCLAP::CmdLine cmd("specMACS camserver", ' ', "0.1");
        TCLAP::MultiArg<string> extraConfigDirs("C","configdirs","Extra folder to look for config files",false,"dirname",cmd);

        cmd.parse(argc, argv);

        for(auto & dirname: extraConfigDirs.getValue()) {
            cout << "found extra config dir: " << dirname << endl;
            CONFIGMANAGER.addConfigDir(dirname);
        }

    } catch (TCLAP::ArgException &e) {
        std::cerr << "error: " << e.error() << " for arg " << e.argId() << std::endl;
        return -1;
    }

    string address;
    {
        NetworkHelper netHelper;
        auto addresses = netHelper.getIpAddresses();
        if(addresses.size() < 1) {
            cerr << "WARNING: found no public IP address! The program will most certainly not work correctly!" << endl;
        } else {
            if(CONFIGMANAGER["publish_address"] != Json::Value::null) {
                string wantedAddress = CONFIGMANAGER["publish_address"].asString();
                for (auto addr : addresses) {
                    if (addr == wantedAddress) {
                        address = addr;
                    }
                }
                if (address.empty()) {
                    cerr << wantedAddress << " not available!" << endl;
                    return -1;
                }
            } else {
                address = *(addresses.begin());
                if(addresses.size() > 1) {
                    cerr << "WARNING: found more than one IP address! Maybe you will experience connection problems!" << endl;
                }
            }
            cout << "INFO: using IP address " << address << endl;
        }
    }
    string rocBroker(SSTR("tcp://", CONFIGMANAGER["broker"].asString(), ":5555"));
    string rocLogBroker(SSTR("tcp://", CONFIGMANAGER["broker"].asString(), ":5556"));
    StorageManager storageManager;
    for (auto & location: CONFIGMANAGER["storageLocations"]) {
        storageManager.addStorageLocation(location.asString());
    }
    keepRunning = true;
#ifdef HAVE_WINDOWS
    SetConsoleCtrlHandler( (PHANDLER_ROUTINE) CtrlHandler, TRUE );
#endif
#ifdef HAVE_LINUX
    struct sigaction sigIntHandler;
    sigIntHandler.sa_handler = SignalHandler;
    sigemptyset(&sigIntHandler.sa_mask);
    sigIntHandler.sa_flags = 0;
    sigaction(SIGINT, &sigIntHandler, NULL);
#endif
    cout << "console handler has been set!" << endl;
    thread outputThread;
    thread logoutputThread;
    
    {
        zmq::context_t ctx(1);
        cout << "zmq context initialized" << endl;
        {
            mutex logUpMutex;
            condition_variable logUpCondition;
            logoutputThread = thread(bind(pubsubproxy,
                                          &ctx,
                                          "inproc://log",
                                          rocLogBroker,
                                          true,
                                          &logUpCondition,
                                          100));
            unique_lock<mutex> logUpLock(logUpMutex);
            cout << "waiting for log output to come up!" << endl;
            logUpCondition.wait(logUpLock);
        }
        cout << "initializing main logger" << endl;
        RocLogger logger(ctx, "inproc://log", "camserver", "main");
        cout << "sleep a little bit..." << endl;
        this_thread::sleep_for(chrono::milliseconds(200)); /* wait a little bit for logging to set up */
        cout << "yawn ... already that late? ... but now I'm awake." << endl;
        logger.debug("camserver starting...");
        cout << "camserver starting..." << endl;
        
        outputThread = thread(bind(pubsubproxy, &ctx, "inproc://output", "tcp://*:6666", false, nullptr, 100));
        
        zmq::socket_t control(ctx, ZMQ_PUB);
        int linger =  0;
        control.setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
        control.bind("inproc://control");
        
        const Json::Value cams = CONFIGMANAGER["cameras"];
        cout << "found cams: " << cams << endl;
        vector<ImageGrabber*> grabbers;
        vector<CamManager*> camManagers;
        vector<Extractor*> extractors;

        string sAutocaptureUUID = uuid4();

        for(unsigned int i=0; i<cams.size(); i++) {
            string name = cams[i]["name"].asString();
            string type = cams[i]["type"].asString();
            string source = cams[i]["source"].asString();

            string local_endpoint = SSTR("inproc://", name, "_raw");

            grabbers.push_back(new ImageGrabber(name, type, source, ctx, rocBroker));
        
            grabbers.back()->addEndpoint(local_endpoint);

            auto publish_to = cams[i]["publish_to"];
            if (!publish_to.empty()) {
                cout << name << " will also be published to " << publish_to << endl;
                grabbers.back()->addEndpoint(publish_to.asString());
            }

            camManagers.push_back(new CamManager(ctx,
                                             rocBroker,
                                             SSTR(name, "CamManager"),
                                             {local_endpoint, name},
                                             storageManager));

            extractors.push_back(new StatsExtractor(ctx,
                                                {local_endpoint, name},
                                                {"inproc://log", name},
                                                name));

            logger.debug(SSTR(name, " size: ", grabbers.back()->getWidth(), "x",
                         grabbers.back()->getHeight(), ", ", grabbers.back()->getBytesPerPixel()));

            grabbers.back()->startServing();
            extractors.back()->startServing();

#ifdef HAVE_LINUX
            auto encode_to = cams[i]["encode_to"];
            if (!encode_to.empty()) {
                cout << name << " will also be encoded to " << encode_to << endl;
                camManagers.back()->addExtractor("FFMpeg", {"FFMpeg", encode_to});
            }
#endif

            /* auto start capture */
            const Json::Value autocapture = cams[i]["autocapture"];
            if(autocapture != Json::Value::null) {
                const Json::Value enabled = autocapture["enabled"];
                if(enabled != Json::Value::null && enabled.asBool() == true) {
                    cout << "autocapture is on for " << name << endl;
                    cout << autocapture << endl;
                    Json::Value metaData(Json::objectValue);
                    if(autocapture["metaData"] != Json::Value::null) {
                        metaData = autocapture["metaData"];
                    }
                    metaData["captureId"] = sAutocaptureUUID;
                    int fileCycling = -1;
                    if(autocapture["fileCycling"] != Json::Value::null) {
                        fileCycling = autocapture["fileCycling"].asInt();
                    }
                    bool binning = false;
                    if(autocapture["binning"] != Json::Value::null) {
                        binning = autocapture["binning"].asBool();
                    }
                    double wantedFPS = 0;
                    if(autocapture["wantedFPS"] != Json::Value::null) {
                        wantedFPS = autocapture["wantedFPS"].asDouble();
                    }
                    cout << "metaData: " << metaData << endl;
                    cout << "fileCycling: " << fileCycling << endl;
                    cout << "binning: " << binning << endl;
                    cout << "wantedFPS: " << wantedFPS << endl;
                    camManagers.back()->addExtractor("fileStorage", {"guiFileStorage",
                                                                     -1,
                                                                     Json::FastWriter().write(metaData),
                                                                     fileCycling,
                                                                     false,
                                                                     binning,
                                                                     wantedFPS});
                }
            }
        }

        list<InfoServer> infoServers;
        for(auto camera: cams) {
            infoServers.emplace_back(ctx, rocBroker, SSTR("camInfo_", camera["name"].asString()), address);
        }
        /*
        ZmqPubExtractor exVnir400(ctx,
                               {"inproc://vnir_raw", VNIR_NAME},
                               {"inproc://output", "vnir400"},
                               {0, 400, vnir.getWidth(), 1});
        ZmqPubExtractor exVnir200(ctx,
                               {"inproc://vnir_raw", VNIR_NAME},
                               {"inproc://output", "vnir200"},
                               {0, 200, vnir.getWidth(), 1});
                               
        ZmqPubExtractor exSwir100(ctx,
                               {"inproc://swir_raw", SWIR_NAME},
                               {"inproc://output", "swir100"},
                               {0, 100, swir.getWidth(), 1});
        */

        
        /*
        exVnir400.startServing();
        exVnir200.startServing();
        exSwir100.startServing();
        */
        
        while(keepRunning) {
            this_thread::sleep_for(chrono::milliseconds(200));
            sendControlMessage(control, "ping");
        }

        sendControlMessage(control, "stop");
        
        /*
        exVnir400.stopServing();
        exVnir200.stopServing();
        exSwir100.stopServing();
        */

        for(auto & extractor: extractors) {
            extractor->stopServing();
            delete extractor;
        }
        for(auto & grabber: grabbers) {
            grabber->stopServing();
            delete grabber;
        }
        for(auto & camManager: camManagers) {
            delete camManager;
        }
        
        logger.debug("camserver finished.");
    }
    cout << "join output threads" << endl;
    outputThread.join();
    cout << "join logoutput threads" << endl;
    logoutputThread.join();
    cout << "join done" << endl;
    return 0;
}