FileStorageExtractor.cpp

Go to the documentation of this file.

#include <FileStorageExtractor.h>
#include <utils.h>
#include <stringtricks.h>
#include <ConfigManager.h>

#include <config.h>

#include <cstring>
#include <cstdio>
#ifdef HAVE_WINDOWS
#define POPEN _popen
#define PCLOSE _pclose
#define POPEN_MODE_WRITE "wb"
#endif
#ifdef HAVE_LINUX
#define POPEN popen
#define PCLOSE pclose
#define POPEN_MODE_WRITE "w"
#endif

#include <iostream>
#include <sstream>

using namespace std;

FileStorageExtractor::FileStorageExtractor(zmq::context_t & _ctx,
                                           const PubSubEndpoint & _source,
                                           const std::string & _identifier,
                                           StorageManager & _storageManager,
                                           int _maxFrames,
                                           const std::string & _metaData,
                                           int _framesPerFileCycle,
                                           bool _cumulative,
                                           bool _binning,
                                           double _wantedFPS) :
        Extractor(_ctx, _source, _wantedFPS),
        s_identifier(_identifier),
        width(0),
        height(0),
        bytesPerPixel(0),
        framesWritten(0),
        totalFramesWritten(0),
        maxFrames(_maxFrames),
        framesPerFileCycle(_framesPerFileCycle),
        cumulative(_cumulative),
        binning(_binning),
        metaData(_metaData),
        error(false),
        accLinear(nullptr),
        accQuadratic(nullptr),
        tmpdata(nullptr) {
    storageManager = &_storageManager;
}

FileStorageExtractor::~FileStorageExtractor() {

}

std::string FileStorageExtractor::type() {
    return "fileStorage";
}

std::string FileStorageExtractor::identifier() {
    return s_identifier;
}

std::list<std::string> FileStorageExtractor::parameters() {
    return {filename, metaData, keepGoing?"active":"done", SSTR(0), SSTR(totalFramesWritten)};
}

/****** PRIVATE *****/

void FileStorageExtractor::extractSetup() {
    openFile();
}

void FileStorageExtractor::extractImpl(unsigned int _width,
                                       unsigned int _height,
                                       unsigned int _bytesPerPixel,
                                       uint64_t highResTime,
                                       uint64_t wallTime,
                                       zmq::message_t * sourceImage) {
    if(binning) {
        _height /=2;
    }
    if(wallTimes.size() == 0) {
        width = _width;
        height = _height;
        bytesPerPixel = _bytesPerPixel;
        if(binning) {
            tmpdata = new uint16_t[width*height];
        }
    } else {
        if(width != _width || height != _height || bytesPerPixel != _bytesPerPixel) {
            cout << "ERROR: image size changed during capture!" << endl;
            error = true;
        }
    }
    if(!error) {
        highResTimes.push_back(highResTime);
        wallTimes.push_back(wallTime);
        if(!cumulative) {
            if(binning) {
                // reduce data
                uint16_t * data = (uint16_t*)sourceImage->data();
                for(unsigned int i=0; i<height; ++i) {
                    for(unsigned int j=0; j<width; ++j) {
                        tmpdata[(i*width)+j] = data[((2*i)*width)+j] + data[(((2*i)+1)*width)+j];
                    }
                }
                size_t actuallyWritten = fwrite(tmpdata, 1, sourceImage->size()/2, fil);
                if(actuallyWritten != sourceImage->size()/2) {
                    cerr << "could only write " << actuallyWritten << " bytes of " << sourceImage->size()/2 << " bytes" << endl;
                    perror("WRITE ERROR");
                }
            } else {
                size_t actuallyWritten = fwrite(sourceImage->data(), 1, sourceImage->size(), fil);
                if(actuallyWritten != sourceImage->size()) {
                    cerr << "could only write " << actuallyWritten << " bytes of " << sourceImage->size() << " bytes" << endl;
                    perror("WRITE ERROR");
                }
            }
        } else {
            unsigned int i;
            unsigned int count = width * height;
            float element;
            if(accLinear == nullptr) {
                accLinear = new float[count]();
            }
            if(accQuadratic == nullptr) {
                accQuadratic = new float[count]();
            }
            switch(_bytesPerPixel) {
                case 1:
                    {
                        uint8_t * data = static_cast<uint8_t*>(sourceImage->data());
                        for(i = 0; i < count; ++i) {
                            element = (float)(data[i]);
                            accLinear[i] += element;
                            accQuadratic[i] += element*element;
                        }
                    }
                    break;
                case 2:
                    {
                        uint16_t * data = static_cast<uint16_t*>(sourceImage->data());
                        for(i = 0; i < count; ++i) {
                            element = (float)(data[i]);
                            accLinear[i] += element;
                            accQuadratic[i] += element*element;
                        }
                    }
                    break;
                case 4:
                    {
                        uint32_t * data = static_cast<uint32_t*>(sourceImage->data());
                        for(i = 0; i < count; ++i) {
                            element = (float)(data[i]);
                            accLinear[i] += element;
                            accQuadratic[i] += element*element;
                        }
                    }
                    break;
                default:
                    cout << "ERROR: " << _bytesPerPixel << " bytesPerPixel are not implemented!" << endl;
                    error = true;
            }
        }
    }
    /* minimal implementation MUST delete sourceImage! */
    delete sourceImage;
    ++framesWritten;
    ++totalFramesWritten;
    if(maxFrames >= 0 && totalFramesWritten >= maxFrames) {
        keepGoing = false;
        return;
    }
    if(framesPerFileCycle > 0 && !cumulative && framesWritten >= framesPerFileCycle) {
        //cycle files
        closeFile();
        framesWritten = 0;
        wallTimes.clear();
        highResTimes.clear();
        openFile();
    }
}

void FileStorageExtractor::extractTeardown() {
    if(cumulative && accLinear != nullptr) {
        size_t size = width * height * sizeof(float);
        size_t actuallyWritten = fwrite((void*)accLinear, 1, size, fil);
        if(actuallyWritten != size) {
            cerr << "could only write " << actuallyWritten << " bytes of " << size << " bytes" << endl;
            perror("WRITE ERROR");
        }
        actuallyWritten = fwrite((void*)accQuadratic, 1, size, fil);
        if(actuallyWritten != size) {
            cerr << "could only write " << actuallyWritten << " bytes of " << size << " bytes" << endl;
            perror("WRITE ERROR");
        }
        delete[] accLinear;
        delete[] accQuadratic;
    }
    if(tmpdata != nullptr) delete[] tmpdata;
    closeFile();
    {
        unique_lock<mutex> _(lastHeaderWriterThreadMutex);
        if(lastHeaderWriterThread != nullptr && lastHeaderWriterThread->joinable()) {
            lastHeaderWriterThread->join();
        }
    }
}

void FileStorageExtractor::openFile() {
    std::string storageDir = storageManager->getPreferredStorageLocation();
    std::string filename_tmp = SSTR(storageDir, "/tmp_recordXXXXXX");
#ifdef HAVE_WINDOWS
    strcpy_s(filename, sizeof(filename), filename_tmp.c_str());
#else
    strcpy(filename, filename_tmp.c_str());
#endif
    int fd = mkstemp(filename);
#ifdef HAVE_WINDOWS
    fil = _fdopen(fd, "wb");
#else
    fil = fdopen(fd, "wb");
#endif
    cout << "opened file " << filename << endl;
}

void FileStorageExtractor::closeFile() {
    cout << "start closing " << fil << endl;
    fclose(fil);
    fil = nullptr;
    cout << "closed file " << filename << endl;
    if(error) {
        cout << "there was an error during file creation, removing " << filename << "!" << endl;
        remove(filename);
    } else if(framesWritten == 0) {
        cout << "no frames written in current cycle, removing " << filename << "!" << endl;
        remove(filename);
    } else {
        stringstream sswalltimes;
        bool first = true;
        for(const auto & t: wallTimes) {
            if(first) {
                sswalltimes << t;
                first = false;
            } else {
                sswalltimes << "," << t;
            }
        }
        stringstream sshighrestimes;
        first = true;
        for(const auto & t: highResTimes) {
            if(first) {
                sshighrestimes << t;
                first = false;
            } else {
                sshighrestimes << "," << t;
            }
        }
        FILE * pPipe;
        if((pPipe = POPEN("python -m runmacs.spec.io.streamed_header", POPEN_MODE_WRITE)) != nullptr) {
            fwriteString(pPipe, "{");
            fwriteString(pPipe, SSTR("\"original_raw_filename\":\"", filename, "\","));
            fwriteString(pPipe, SSTR("\"source_name\":\"", source.topic, "\","));
            fwriteString(pPipe, "\"sw_vers\":\"" DAQ_SW_VERSION "\",");
            fwriteString(pPipe, SSTR("\"wall_times\":[", sswalltimes.str(), "],"));
            fwriteString(pPipe, SSTR("\"highRes_times\":[", sshighrestimes.str(), "],"));
            fwriteString(pPipe, SSTR("\"specbin\":", (binning?"2":"1"), ","));
            fwriteString(pPipe, "\"spatbin\":1,");
            fwriteString(pPipe, SSTR("\"width\":", (int)width, ","));
            fwriteString(pPipe, SSTR("\"height\":", (int)height*(binning?2:1), ","));
            if(cumulative) {
                fwriteString(pPipe, "\"format\":\"f\",");
                fwriteString(pPipe, "\"cumulative\":true,");
            } else {
                fwriteString(pPipe, "\"format\":\"H\",");
                fwriteString(pPipe, "\"cumulative\":false,");
            }
            fwriteString(pPipe, SSTR("\"sensorControlServer\":\"", CONFIGMANAGER["daq_broker"].asString(), "\","));
            fwriteString(pPipe, SSTR("\"metaData\":", metaData, ""));
            fwriteString(pPipe, "}");
            /* headerwriter should be closed asynchronously to allow for the next file to be written */
            {
                unique_lock<mutex> _(lastHeaderWriterThreadMutex);
                thread * prevHeaderWriterThread = lastHeaderWriterThread;
                lastHeaderWriterThread = new thread([=](){
                            (void) PCLOSE(pPipe);
                            if(prevHeaderWriterThread != nullptr && prevHeaderWriterThread->joinable()) {
                                prevHeaderWriterThread->join();
                            }
                        });
            }
            //int returnCode = PCLOSE(pPipe);
        } else {
            cout << "could not call header-writer" << endl;
        }
    }
}