FileReader.hpp

Go to the documentation of this file.

 
#ifndef H_POTTU_FILEREADER
#define H_POTTU_FILEREADER
 
 
#include "TDRFile.hpp"
//#include "DataPacketRawItems.hpp"
#include "dataitem.hpp"
#include "ContextBase.hpp"
#include "DataSource.hpp"
 
 
#include <fmt/base.h>
 
#include <deque>
#include <vector>
#include <type_traits>
#include <cassert>
#include <memory>
#include <chrono>
 
 
namespace pottu {
 
    class FileReader : public DataSource {
    public:
 
    FileReader( uint32_t blocksAtOnce = 1 )
        : _ctxh( ContextBase::getActive().createHandle("FileReader") ),
          _blocksAtOnce(blocksAtOnce)
        {
        if( blocksAtOnce < 1 )
            throw std::runtime_error( "FileReader(): blocksAtOnce must be at least 1!" );
        }
 
    void addFile( const std::string &filename ) {
        auto f = std::make_unique<TDRFile>( filename );
        f->open();
        f->close();
        _totalsize += f->getFilesize();
        _ctxh->logInfo( fmt::format( "File {} added", f->getPath() ) );
        _filesWaiting.push_back( std::move(f) );
    }
 
    
    virtual void process( std::vector<dataitem_t> &dp ) {
        static std::vector<uint8_t> buf;
        auto proci = _ctxh->processInstance();
        dp.clear();
 
        for( uint32_t i = 0; i < _blocksAtOnce; ++i ) {
        
        if( !_currentFile ) {
            _ctxh->logInfo( fmt::format( "No file open. Trying to open next one if any." ) );
            if( _filesWaiting.size() ) {
            _currentFile = std::move( _filesWaiting.front() );
            _filesWaiting.pop_front();
            } else {
            _ctxh->logInfo( fmt::format( "All files processed." ) );
            return;
            }
            _currentFile->open();
            _ctxh->logInfo( fmt::format( "Starting to process file {}.", _currentFile->getPath() ) );
        }
        buf.resize( _currentFile->getBlocksize() );
        
        // Reading next block. len is length of usefull data
        // excluding the header. buf length is however always
        // blocksize and contains the header.
        size_t len = _currentFile->readBlock(buf.data());
        //fmt::print( "sequence: {}, len: {}\n", _currentFile->getLastHeader().sequence, len );
        if( len % 8 != 0 ) {
            /*
            _ctxh->logWarning( fmt::format( "Header size not multiple of 8. file: {}, sequence: {}, len: {}",
                          _currentFile->getPath(), _currentFile->getLastHeader().sequence, len ) );
            */
            len = (len / 8)*8;
        }
 
        if( len ) {
            // Making room for new data at the end of vector
            // data. Then copying data there with memcpy (cannot
            // use insert since the alignment of blockbuffer is
            // wrong).
            auto firstNew = dp.size();
            dp.resize( dp.size() + len/8 );
 
            /*
            fmt::print( "{:12} {:12} {:12}\n",
                dp.size(), &*dp.end()-(dp.data()+firstNew), len/8.0 );
            */
            std::memcpy( dp.data()+firstNew, buf.data()+24, len );
            ++_blocksRead;
            _totalBytesRead += _currentFile->getBlocksize();
        
            // Swapping dataitems if needed
            if( _currentFile->isWordSwapRequired() ) {
            for( auto it = dp.begin()+firstNew; it != dp.end(); ++it )
                *it = it->asSwapped();
            }
        }
        
        if( _currentFile->isEof() ) {
            _ctxh->logInfo( fmt::format( "File {} processed", _currentFile->getPath() ) );
            _totalBytesReadFromFile += _currentFile->getGZFile().getPosition();
            _currentFile->close();
            _filesProcessed.push_back( std::move(_currentFile) );
        }
        
        }
 
 
        if( _reportProgress ) {
        auto t = std::chrono::steady_clock::now();
        if( t-_lastProgressReport > _maxReportInterval ) {
            double dt = (std::chrono::duration<double>( t-_lastProgressReport ) ).count();
            double progress = getProgress();
            double estLeftMinutes = (1.0-progress)/( (progress-_lastProgress)/dt ) / 60.0;
            _ctxh->logInfo( fmt::format( "Progress: {:>3}/{:<3}, {:6.3f} %, est. {:6.2f} min",
                         _filesProcessed.size()+1,
                         _filesProcessed.size()+1+_filesWaiting.size(),
                         progress*100,
                         estLeftMinutes ) );
            _lastProgressReport = t;
            _lastProgress = progress;
        }
        }
        
    }
 
    virtual bool allDone() const noexcept { return !_currentFile && _filesWaiting.empty(); }
 
    uint64_t getBlocksRead() const { return _blocksRead; }
 
    uint64_t getTotalBytesRead() const { return _totalBytesRead; } 
 
    uint64_t getTotalBytesReadFromFile() const {
        return _totalBytesReadFromFile + (_currentFile ? _currentFile->getGZFile().getPosition() : 0 );
    }
 
    double getProgress() const noexcept {
        return getTotalBytesReadFromFile()/double(_totalsize);
    }
 
    void setProgressReport( bool state=true ) noexcept {
        _reportProgress = state;
    }
 
    bool getProgressReport() const noexcept { return _reportProgress; }
 
    
    private:
    
    ContextHandle                *_ctxh;
 
    uint32_t                      _blocksAtOnce;
    
    std::unique_ptr<TDRFile>              _currentFile;
    std::deque<std::unique_ptr<TDRFile>>  _filesWaiting;
    std::deque<std::unique_ptr<TDRFile>>  _filesProcessed;
 
 
    uint64_t                      _totalsize{0};
    uint64_t                      _blocksRead{0};
    uint64_t                      _totalBytesRead{0};
    uint64_t                      _totalBytesReadFromFile{0};
 
    bool                                  _reportProgress{true};
    std::chrono::steady_clock::duration   _maxReportInterval{std::chrono::seconds(2)};
    std::chrono::steady_clock::time_point _lastProgressReport{std::chrono::steady_clock::now()};
    double                                _lastProgress{0};
    };
 
 
}
 
    
#endif