Tape.h

Go to the documentation of this file.

/*
 * Copyright (C) 2012 by
 *   MetraLabs GmbH (MLAB), GERMANY
 * and
 *   Neuroinformatics and Cognitive Robotics Labs (NICR) at TU Ilmenau, GERMANY
 * All rights reserved.
 *
 * Contact: info@mira-project.org
 *
 * Commercial Usage:
 *   Licensees holding valid commercial licenses may use this file in
 *   accordance with the commercial license agreement provided with the
 *   software or, alternatively, in accordance with the terms contained in
 *   a written agreement between you and MLAB or NICR.
 *
 * GNU General Public License Usage:
 *   Alternatively, this file may be used under the terms of the GNU
 *   General Public License version 3.0 as published by the Free Software
 *   Foundation and appearing in the file LICENSE.GPL3 included in the
 *   packaging of this file. Please review the following information to
 *   ensure the GNU General Public License version 3.0 requirements will be
 *   met: http://www.gnu.org/copyleft/gpl.html.
 *   Alternatively you may (at your option) use any later version of the GNU
 *   General Public License if such license has been publicly approved by
 *   MLAB and NICR (or its successors, if any).
 *
 * IN NO EVENT SHALL "MLAB" OR "NICR" BE LIABLE TO ANY PARTY FOR DIRECT,
 * INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF
 * THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF "MLAB" OR
 * "NICR" HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * "MLAB" AND "NICR" SPECIFICALLY DISCLAIM ANY WARRANTIES, INCLUDING,
 * BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
 * FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND "MLAB" AND "NICR" HAVE NO OBLIGATION TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS OR MODIFICATIONS.
 */

#ifndef _MIRA_TAPE_H_
#define _MIRA_TAPE_H_

#ifndef Q_MOC_RUN
#include <boost/optional.hpp>
#endif

#include <utils/Path.h>
#include <fw/Framework.h>

#include <platform/Platform.h>

namespace mira {


class MIRA_FRAMEWORK_EXPORT Tape
{
public:
    static const uint32 sHeaderSize;
    // size of a message block packet
    static const uint32 sMessageBlockSize;

    enum OpenMode
    {
        READ,
        WRITE,
        INFO,   
    };

    enum HeaderType
    {
        MESSAGEBLOCK = 0x0A,
        CHANNELINFO = 0x0B,
        MESSAGE = 0x0C,
        INDEX = 0x0D
    };

    struct Header
    {
        char type;      
        uint32 size;    
    };

    struct MessageIndex
    {
        uint64 block;           
        uint64 offset;          
        Duration timeOffset;    
    };

    struct ChannelInfo
    {
        typedef std::vector<MessageIndex> MessageVector;
        uint64 offset;              
        uint64 offsetToIndexTable;  
        Duration firstMessageOffset;
        Duration lastMessageOffset; 
        std::string name;           
        std::string type;           
        TypeMetaPtr meta;           
        MetaTypeDatabase metaDB;    
        MessageVector messages;     
        uint64 dataSize;            
    };

    struct FileInfo
    {
        std::fstream file;          
        Time start;                 
        Duration timezoneOffset;    
        std::string filename;       
        uint32 version;             
        uint32 nrBlocks;            
        uint32 nrChannels;          
        uint64 offsetToFirstInfo;   
    };

    struct MessageBlock
    {
        uint64 offset;                  
        uint32 nrMessages;              
        uint32 size;                    
        Duration firstMessageOffset;    
        Duration lastMessageOffset;     
        Buffer<uint8> buffer;           
    };

    struct Message
    {
        bool compressed;        
        std::string name;       
        std::string frameID;    
        uint32 sequenceID;      
        Buffer<uint8> data;     
        uint32 uncompressedSize;
    };

    typedef std::map<std::string, ChannelInfo> ChannelMap;
    typedef std::map<uint64, MessageBlock> MessageBlockMap;
    typedef std::multimap<Duration, Message> MessageMap;

    Tape() :
        mIsOpen(false),
        mWaitForAlteredStartTime(false),
        mMaxMessageBlockSize(64 * 1024* 1024),
        mSortWindowSize(Duration::seconds(2))
    {}

    ~Tape();

    void open(const Path& file, OpenMode mode);

    void close();

    void repair(const Path& file, const Path& outFile);

    void waitForAlteredStartTime();

    void alterStartTime(const Time& startTime);

    template <typename T>
    MIRA_DEPRECATED("Please use int compressionLevel(=0/-1) instead of bool compress(=false/true)",
    std::size_t write(ChannelRead<T> value, bool compress,
                      TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase()))
    {
        if (!mIsOpen || mMode != WRITE)
            MIRA_THROW(XIO, "Tape is not opened in write mode");
        return write(value.getChannelID(), value.getTypename(),
                     value->timestamp, value.readSerialized(),
                     (compress ? -1 : 0), meta, metaDB);
    }

    template <typename T>
    std::size_t write(ChannelRead<T> value, int compressionLevel = 0,
                      TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase())
    {
        if (!mIsOpen || mMode != WRITE)
            MIRA_THROW(XIO, "Tape is not opened in write mode");
        return write(value.getChannelID(), value.getTypename(),
                     value->timestamp, value.readSerialized(),
                     compressionLevel, meta, metaDB);
    }

    template <typename T>
    MIRA_DEPRECATED("Please use int compressionLevel(=0/-1) instead of bool compress(=false/true)",
    std::size_t write(ChannelRead<T> value,
                      std::list<BinarySerializerCodecPtr>& codecs,
                      bool compress, TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase()))
    {
        if (!mIsOpen || mMode != WRITE)
            MIRA_THROW(XIO, "Tape is not opened in write mode");
        return write(value.getChannelID(), value.getTypename(),
                     value->timestamp, value.readSerialized(codecs),
                     (compress ? -1 : 0), meta, metaDB);
    }

    template <typename T>
    std::size_t write(ChannelRead<T> value,
                      std::list<BinarySerializerCodecPtr>& codecs,
                      int compressionLevel = 0, TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase())
    {
        if (!mIsOpen || mMode != WRITE)
            MIRA_THROW(XIO, "Tape is not opened in write mode");
        return write(value.getChannelID(), value.getTypename(),
                     value->timestamp, value.readSerialized(codecs),
                     compressionLevel, meta, metaDB);
    }

    template <typename T>
    MIRA_DEPRECATED("Please use int compressionLevel(=0/-1) instead of bool compress(=false/true)",
    std::size_t write(const std::string& channelID, const Stamped<T>& value,
                      bool compress, TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase()))
    {
        Buffer<uint8> buffer;
        BinaryBufferSerializer s(&buffer);
        s.serialize(value.value(),false);
        return write(channelID, typeName<T>(), value.timestamp,
                     value.frameID, value.sequenceID,
                     buffer, (compress ? -1 : 0), meta, metaDB);
    }

    template <typename T>
    std::size_t write(const std::string& channelID, const Stamped<T>& value,
                      int compressionLevel = 0, TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase())
    {
        Buffer<uint8> buffer;
        BinaryBufferSerializer s(&buffer);
        s.serialize(value.value(),false);
        return write(channelID, typeName<T>(), value.timestamp,
                     value.frameID, value.sequenceID,
                     buffer, compressionLevel, meta, metaDB);
    }

    MIRA_DEPRECATED("Please use int compressionLevel(=0/-1) instead of bool compress(=false/true)",
    std::size_t write(const std::string& channelID, const std::string& typeName,
                      const StampedHeader& header,
                      const Buffer<uint8>& data, bool compress,
                      TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase()))
    {
        return write(channelID, typeName, header.timestamp, header.frameID,
                     header.sequenceID, data, (compress ? -1 : 0), meta, metaDB);
    }

    std::size_t write(const std::string& channelID, const std::string& typeName,
                      const StampedHeader& header,
                      const Buffer<uint8>& data, int compressionLevel = 0,
                      TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase())
    {
        return write(channelID, typeName, header.timestamp, header.frameID,
                     header.sequenceID, data, compressionLevel, meta, metaDB);
    }

    MIRA_DEPRECATED("Please use int compressionLevel(=0/-1) instead of bool compress(=false/true)",
    std::size_t write(const std::string& channelID, const std::string& typeName,
                      const Time& time, const std::string& frameID, uint32 sequenceID,
                      const Buffer<uint8>& data, bool compress,
                      TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase()))
    {
        return write(channelID, typeName, time, frameID, sequenceID,
                     data, (compress ? -1 : 0), meta, metaDB);
    }

    std::size_t write(const std::string& channelID, const std::string& typeName,
                      const Time& time, const std::string& frameID, uint32 sequenceID,
                      const Buffer<uint8>& data, int compressionLevel = 0,
                      TypeMetaPtr meta = TypeMetaPtr(),
                      const MetaTypeDatabase& metaDB = MetaTypeDatabase());

    void setMaxMessageBlockSize(uint32 size)
    {
        mMaxMessageBlockSize = size;
    }

    void setSortWindowSize(const Duration& interval)
    {
        mSortWindowSize = interval;
    }

    void setStartTime(Time time)
    {
        mFile.start = time;
        mFile.timezoneOffset = mFile.start.toLocal() - mFile.start;
    }

    Time getStartTime() const
    {
        if (!mIsOpen)
            MIRA_THROW(XIO, "Tape is not opened.");
        return mFile.start;
    }

    Time getLocalStartTime() const
    {
        if (!mIsOpen)
            MIRA_THROW(XIO, "Tape is not opened.");
        return mFile.start + mFile.timezoneOffset;
    }

    Time getEndTime() const
    {
        if (!mIsOpen)
            MIRA_THROW(XIO, "Tape is not opened.");
        if (mMessageBlocks.size()==0)
            return Time::unixEpoch();
        return mFile.start + mMessageBlocks.rbegin()->second.lastMessageOffset;
    }

    Duration getTZOffset() const
    {
        if (!mIsOpen)
            MIRA_THROW(XIO, "Tape is not opened.");
        return mFile.timezoneOffset;
    }

    const Tape::ChannelMap& getChannels() const
    {
        if (!mIsOpen)
            MIRA_THROW(XIO, "Tape is not opened.");
        return mChannels;
    }

    const Tape::MessageBlockMap& getMessageBlocks() const
    {
        if (!mIsOpen)
            MIRA_THROW(XIO, "Tape is not opened.");
        return mMessageBlocks;
    }

    uint32 getVersion() const
    {
        if (!mIsOpen)
            MIRA_THROW(XIO, "Tape is not opened.");
        return mFile.version;
    }

    static uint32 getCurrentVersion();

    void readMessage(const MessageIndex& index, std::string& frameID,
                     uint32& sequenceID, Buffer<uint8>& oData, Duration& oTime);

    void readMessage(const MessageIndex& index, std::string& frameID,
                     uint32& sequenceID, Buffer<uint8>& oData, Duration& oTime,
                     bool& oCompressed);


protected:

    uint64 getFileHeaderSize() const;
    void openInfo(const Path& file);
    void openRead(const Path& file);
    void closeRead();
    void openWrite(const Path& file);
    void closeWrite();
    void openRepair(const Path& file);

    void write(Duration time, const Message& message);

    void readFileHeader();
    void readChannelInfo();
    void readMessageBlocks();
    void readMessageBlock(uint64 offset);
    void writeFileHeader();
    Tape::Header readHeader();
    void writeHeader(HeaderType type, uint32 size);
    void writeChannelInfo(ChannelInfo& info);
    void startMessageBlock(Duration time);
    void finishMessageBlock();

protected:

    bool mIsOpen;
    bool mWaitForAlteredStartTime;
    OpenMode mMode;
    ChannelMap mChannels;
    std::set<std::string> mWrittenChannels;
    boost::optional<MessageBlock> mCurrentMessageBlock;
    boost::mutex mMutex;
    boost::mutex mMessageMutex;
    MessageBlockMap mMessageBlocks; 
    FileInfo mFile;                 
    uint64 mLastInfo;               
    uint32 mMaxMessageBlockSize;    

    MessageMap mMessages;           
    Duration mSortWindowSize;       
};


}

#endif