source: trunk/Mars/mcore/zfits.h@ 16561

/*
 * zfits.h
 *
 *  Created on: May 16, 2013
 *      Author: lyard
 */

#ifndef MARS_ZFITS
#define MARS_ZFITS

#include <stdexcept>

#include "fits.h"
#include "huffman.h"


#ifndef __MARS__
namespace std
{
#endif

class zfits : public fits
{
public:

    // Basic constructor
    zfits(const string& fname, const string& tableName="",
          bool force=false) : fits(fname, tableName, force),
        fBuffer(0),
        fTransposedBuffer(0),
        fCompressedBuffer(0),
        fNumTiles(0),
        fNumRowsPerTile(0),
        fCurrentRow(-1),
        fHeapOff(0),
        fTileSize(0)
    {
        InitCompressionReading();
    }

    // Alternative contstructor
    zfits(const string& fname, const string& fout, const string& tableName,
          bool force=false) : fits(fname, fout, tableName, force),
              fBuffer(0),
              fTransposedBuffer(0),
              fCompressedBuffer(0),
              fNumTiles(0),
              fNumRowsPerTile(0),
              fCurrentRow(-1),
              fHeapOff(0),
              fTileSize(0)
    {
        InitCompressionReading();
    }

    //  Skip the next row
    bool SkipNextRow()
    {
        if (!fTable.isCompressed)
            return fits::SkipNextRow();

        fRow++;
        return true;
    }
protected:

    //  Stage the requested row to internal buffer
    //  Does NOT return data to users
    virtual void StageRow(size_t row, char* dest)
    {
        if (!fTable.isCompressed)
        {
            fits::StageRow(row, dest);
            return;
        }

        ReadBinaryRow(row, dest);
    }

private:

    // Do what it takes to initialize the compressed structured
    void InitCompressionReading()
    {
        //The constructor may have failed
        if (!good())
            return;

        //Get compressed specific keywords
        fNumTiles       = fTable.isCompressed ? GetInt("NAXIS2") : 0;
        fNumRowsPerTile = fTable.isCompressed ? GetInt("ZTILELEN") : 0;

        //give it some space for uncompressing
        AllocateBuffers();

        //read the file's catalog
        ReadCatalog();
    }

    // Copy decompressed data to location requested by user
    void MoveColumnDataToUserSpace(char* dest, const char* src, const Table::Column& c)
    {
        if (!fTable.isCompressed)
        {
            fits::MoveColumnDataToUserSpace(dest, src, c);
            return;
        }

        memcpy(dest, src, c.num*c.size);
    }

    vector<char> fBuffer;           ///<store the uncompressed rows
    vector<char> fTransposedBuffer; ///<intermediate buffer to transpose the rows
    vector<char> fCompressedBuffer; ///<compressed rows

    size_t fNumTiles;       ///< Total number of tiles
    size_t fNumRowsPerTile; ///< Number of rows per compressed tile
    size_t fCurrentRow;     ///< current row in memory.

    streamoff fHeapOff;     ///< offset from the beginning of the file of the binary data

    vector<vector<pair<int64_t, int64_t> > > fCatalog;///< Catalog, i.e. the main table that points to the compressed data.
    vector<size_t> fTileSize; ///< size in bytes of each compressed tile
    vector<vector<size_t> > fTileOffsets; ///< offset from start of tile of a given compressed column

    void AllocateBuffers()
    {
        if (!fTable.isCompressed)
            return;

        fBuffer.resize(fTable.bytes_per_row*fNumRowsPerTile);

        fTransposedBuffer.resize(fTable.bytes_per_row*fNumRowsPerTile);
        fCompressedBuffer.resize(fTable.bytes_per_row*fNumRowsPerTile + fTable.num_cols); //use a bit more memory for compression flags
    }

    // Read catalog data. I.e. the address of the compressed data inside the heap
    void ReadCatalog()
    {
        if (!fTable.isCompressed)
            return;

        char readBuf[16];
        fCatalog.resize(fNumTiles);

        const streampos catalogStart = tellg();

        //do the actual reading
        for (uint32_t i=0;i<fNumTiles;i++)
            for (uint32_t j=0;j<fTable.num_cols;j++)
            {
                read(readBuf, 2*sizeof(int64_t));

                //swap the bytes
                int64_t tempValues[2] = {0,0};
                revcpy<8>(reinterpret_cast<char*>(tempValues), readBuf, 2);
                if (tempValues[0] < 0 || tempValues[1] < 0)
                {
#ifdef __EXCEPTIONS
                    throw runtime_error("ERROR: negative value in the catalog");
#else
                    gLog << ___err ___ << "ERROR: negative value in the catalog" << endl;
                    return;
#endif
                }
                //add catalog entry
                fCatalog[i].emplace_back(tempValues[0], tempValues[1]);
            }

        //compute the total size of each compressed tile
        fTileSize.resize(fNumTiles);
        fTileOffsets.resize(fNumTiles);
        for (uint32_t i=0;i<fNumTiles;i++)
        {
            fTileSize[i] = 0;
            for (uint32_t j=0;j<fTable.num_cols;j++)
            {
                fTileSize[i] += fCatalog[i][j].first;
                fTileOffsets[i].emplace_back(fCatalog[i][j].second - fCatalog[i][0].second);
            }
        }
        //see if there is a gap before heap data
        fHeapOff = tellg()+fTable.GetHeapShift();

        if (!fCopy.is_open())
            return;

        //write catalog and heap gap to target file
        seekg(catalogStart);

        const size_t catSize = fTable.GetHeapShift() + fTable.total_bytes;

        vector<char> buf(catSize);
        read(buf.data(), catSize);

        fCopy.write(buf.data(), catSize);
        if (!fCopy)
            clear(rdstate()|ios::badbit);
    }
    //overrides fits.h method with empty one
    //work is done in ReadBinaryRow because it requires volatile data from ReadBinaryRow
    virtual void WriteRowToCopyFile(size_t )
    {

    }
    // Compressed versin of the read row
    bool ReadBinaryRow(const size_t &rowNum, char *bufferToRead)
    {
        if (rowNum >= GetNumRows())
            return false;

        const uint32_t requestedTile = rowNum/fNumRowsPerTile;
        const uint32_t currentTile   = fCurrentRow/fNumRowsPerTile;
        const size_t   previousRow   = fCurrentRow;

        fCurrentRow = rowNum;

        //should we read yet another chunk of data ?
        if (requestedTile != currentTile)
        {
            //read yet another chunk from the file
            const int64_t sizeToRead = fTileSize[requestedTile];

            //skip to the beginning of the tile
            seekg(fHeapOff+fCatalog[requestedTile][0].second);
            read(fCompressedBuffer.data(), sizeToRead);

            if (fCurrentRow == previousRow+1 &&
                fCopy.is_open() &&
                fCopy.good())
            {
                fCopy.write(fCompressedBuffer.data(), sizeToRead);
                if (!fCopy)
                    clear(rdstate()|ios::badbit);
            }
            else
                if (fCopy.is_open())
                    clear(rdstate()|ios::badbit);

            const uint32_t thisRoundNumRows = (GetNumRows()<fCurrentRow + fNumRowsPerTile) ? GetNumRows()%fNumRowsPerTile : fNumRowsPerTile;

            //uncompress it
            UncompressBuffer(requestedTile, thisRoundNumRows);

            // pointer to column (source buffer)
            const char *src = fTransposedBuffer.data();

            for (auto it=fTable.sortedCols.begin(); it!=fTable.sortedCols.end(); it++)
            {
                char *buffer = fBuffer.data() + it->offset; // pointer to column (destination buffer)

                switch (it->comp)
                {
                case UNCOMPRESSED:
                case SMOOTHMAN:
                    // regular, "semi-transposed" copy
                    for (char *dest=buffer; dest<buffer+thisRoundNumRows*fTable.bytes_per_row; dest+=fTable.bytes_per_row) // row-by-row
                    {
                        memcpy(dest, src, it->bytes);
                        src += it->bytes;  // next column
                    }
                    break;

                default:
                    // transposed copy
                    for (char *elem=buffer; elem<buffer+it->bytes; elem+=it->size) // element-by-element (arrays)
                    {
                        for (char *dest=elem; dest<elem+thisRoundNumRows*fTable.bytes_per_row; dest+=fTable.bytes_per_row) // row-by-row
                        {
                            memcpy(dest, src, it->size);
                            src += it->size; // next element
                        }
                    }
                    break;
                };
            }
        }

        //Data loaded and uncompressed. Copy it to destination
        memcpy(bufferToRead, fBuffer.data()+fTable.bytes_per_row*(fCurrentRow%fNumRowsPerTile), fTable.bytes_per_row);
        return good();
    }

    // Read a bunch of uncompressed data
    uint32_t UncompressUNCOMPRESSED(char*       dest,
                                    const char* src,
                                    uint32_t    numRows,
                                    uint32_t    sizeOfElems,
                                    uint32_t    numRowElems)
    {
        memcpy(dest, src, numRows*sizeOfElems*numRowElems);
        return numRows*sizeOfElems*numRowElems;
    }

    // Read a bunch of data compressed with the Huffman algorithm
    uint32_t UncompressHUFFMAN(char*       dest,
                               const char* src,
                               uint32_t ,
                               uint32_t    sizeOfElems,
                               uint32_t    numRowElems)
    {
        if (sizeOfElems < 2)
        {
            cout << "Error, Huffman only works on shorts or longer types. (here: " << sizeOfElems << "). Aborting." << endl;
            return -1;
        }

        vector<uint16_t> uncompressed;

        //read compressed sizes (one per row)
        const uint32_t* compressedSizes = reinterpret_cast<const uint32_t*>(src);
        src += sizeof(uint32_t)*numRowElems;

        //uncompress the rows, one by one
        uint32_t sizeWritten = 0;
        for (uint32_t j=0;j<numRowElems;j++)
        {
            Huffman::Decode(reinterpret_cast<const unsigned char*>(src), compressedSizes[j], uncompressed);

            memcpy(dest, uncompressed.data(), uncompressed.size()*sizeof(uint16_t));

            sizeWritten += uncompressed.size()*sizeof(uint16_t);
            dest        += uncompressed.size()*sizeof(uint16_t);
            src         += compressedSizes[j];
        }
        return sizeWritten;
    }

    //Read a bunch of data compressed with the smoothman algorithm
    uint32_t UncompressSMOOTHMAN(int16_t*   dest,
                                 const char* src,
                                 uint32_t numRows,
                                 uint32_t sizeOfElems,
                                 uint32_t numRowElems)
    {
        //call huffman transposed
        const uint32_t sizeWritten = UncompressHUFFMAN(reinterpret_cast<char*>(dest), src, numRowElems, sizeOfElems, numRows);

        //un-do the integer smoothing
        for (uint32_t j=2;j<numRowElems*numRows;j++)
            dest[j] = dest[j] + (dest[j-1]+dest[j-2])/2;

        return sizeWritten;
    }

    // Data has been read from disk. Uncompress it !
    void UncompressBuffer(const uint32_t &catalogCurrentRow, const uint32_t &thisRoundNumRows)
    {
        char *dest = fTransposedBuffer.data();

        //uncompress column by column
        for (uint32_t i=0; i<fTable.sortedCols.size(); i++)
        {
            const fits::Table::Column &col = fTable.sortedCols[i];
            if (col.num == 0)
                continue;

            //get the compression flag
            const int64_t compressedOffset = fTileOffsets[catalogCurrentRow][i];//fCatalog[catalogCurrentRow][i].second - fCatalog[catalogCurrentRow][0].second;
            const char    compressedFlag   = fCompressedBuffer[compressedOffset];

            //#define COMPRESSED_FLAG 0x1
            //#define UNCOMPRESSED_FLAG 0x0

            const char *src = fCompressedBuffer.data()+compressedOffset+1;

            //if this bunch of data is not compressed, modify the compression flag
            const uint32_t compression = compressedFlag==0 ? UNCOMPRESSED : col.comp;
            switch (compression)
            {
                case UNCOMPRESSED:
                    dest += UncompressUNCOMPRESSED(dest, src, thisRoundNumRows, col.size, col.num);
                    break;

                case SMOOTHMAN:
                    dest += UncompressSMOOTHMAN(reinterpret_cast<int16_t*>(dest), src, thisRoundNumRows, col.size, col.num);
                    break;

                default:
                    ;
            }
        }
    }

};//class zfits

#ifndef __MARS__
}; //namespace std
#endif

#endif